HK40030421B - Managing a computing cluster using time interval counters - Google Patents

Description

Use time interval counters to manage the compute cluster

Cross-reference to related applications

This application claims priority to U.S. Application Serial No. 62/579,225, filed October 31, 2017, which is incorporated herein by reference.

Technical Field

This manual pertains to the management of computing clusters.

Background Technology

One method of dataflow computation utilizes a graph-based representation, in which computational components corresponding to nodes (vertices) of a graph are coupled via data flows corresponding to links (directed edges) of the graph (referred to as a “dataflow graph”). Downstream components connected to upstream components via data flow links receive ordered streams of input data elements and process them in the received order to optionally generate one or more corresponding streams of output data elements. A system for performing such graph-based computation is described in prior U.S. Patent 5,966,072, entitled “EXECUTING COMPUTATIONS EXPRESSED AS GRAPHS,” which is incorporated herein by reference. In embodiments relating to the method described in that prior patent, each component is implemented as a process hosted on one of a plurality of typical computer servers. Each computer server may have multiple such component processes active at any given time, and the operating system (e.g., Unix) scheduler shares resources (e.g., processor time and/or processor cores) among the components hosted on that server. In this implementation, data flow between components can be implemented using data communication services (e.g., named pipes, TCP/IP sessions, etc.) of the operating system and the data network connecting the server. A subset of components typically serves as a data source and/or data receiver for the entire computation (e.g., to and/or from data files, database tables, and external data streams). After component processes and data flows are established, for example, through a collaborative process, data flows through the entire computing system, which performs computations represented as graphs, typically governed by the availability of input data at each component and the scheduling of computational resources for each component. Therefore, parallelism can be achieved at least by enabling different components to be executed in parallel by different processes (hosted on the same or different server computers or processor cores). Herein, parallel execution of different components on different paths via a data flow graph is referred to as component parallelism, and parallel execution of different components on different parts of the same path via a data flow graph is referred to as pipeline parallelism.

This approach also supports other forms of parallelism. For example, the input dataset can be partitioned, for instance, based on the values of fields in the records of the dataset, with each part being sent to a component to process a separate copy of the records in that dataset. Such separate copies (or "instances") of the components can execute on a single server computer or a single processor core of a server computer, thus achieving what is referred to herein as data parallelism. The results of the individual components can be merged to form a single data stream or dataset again. The number of computers or processor cores used to execute the instances of the components will be specified by the developer when developing the data flow graph.

Various methods can be used to improve the efficiency of this approach. For example, each instance of a component does not necessarily have to be hosted in its own operating system process; for instance, multiple components can be implemented using a single operating system process (e.g., components that form connected subgraphs of a larger graph).

At least some implementations of the above methods are limited by the efficiency of the processes generated during execution on the underlying computer server. For example, these limitations may be related to the difficulty of reconfiguring running instances of the graph to change the degree of data parallelism, changing the servers hosting various components, and/or balancing the load across different computing resources. Existing graph-based computing systems also suffer from slow startup times, often due to the unnecessary startup of too many processes, wasting significant amounts of memory. Generally, processes begin at the start of graph execution and end when graph execution is complete.

Other systems for distributed computing have been used, in which the entire computation is divided into smaller parts and these parts are distributed from a master computer server to various other (e.g., "slave") computer servers, each of which performs the computation independently and returns its results to the master server. Some of these approaches are called "grid computing." However, such approaches typically rely on the independence of each computation and do not provide mechanisms for transferring data between those parts, or for scheduling and/or ordering the execution of these parts, except for invoking the computation parts via the master computer server. Therefore, such approaches do not provide a direct and efficient solution for hosting computations involving interactions between multiple components.

Another approach to distributed computation on large datasets utilizes the MapReduce framework (e.g., as implemented in Apache systems). Generally, Hadoop has a distributed file system where portions of each named file are distributed. Users specify computations based on two functions: a mapping function executed distributedly on all portions of the named input and a reduction function executed on the output portion of the mapping function. The output of the mapping function is partitioned and stored again in intermediate portions within the distributed file system. The reduction function is then executed distributedly to process these intermediate portions, yielding the result of the entire computation. While computations that can be represented within the MapReduce framework, and whose inputs and outputs can be modified to be stored within the MapReduce framework's file system, can be performed efficiently, many computations are incompatible with this framework and/or not easily adapted to have all their inputs and outputs contained within a distributed file system.

Summary of the Invention

In general, a method for processing state update requests in a distributed data processing system with multiple processing nodes includes using two or more of these processing nodes to process a set of requests, each request in each set being configured to cause a state update at one of these processing nodes and associated with a corresponding time interval among a plurality of time intervals. The set of state update requests includes a first set of requests associated with a first time interval.

The method further includes maintaining a plurality of counters at one of these processing nodes, the plurality of counters including: a working counter that indicates the current time interval and the value of the current time interval in the distributed data processing system; a replication counter that indicates a time interval and the value of the time interval in which all requests associated with the time interval are replicated at several processing nodes among the plurality of processing nodes; and a persistence counter that indicates a time interval in which all requests associated with the time interval are stored in persistent storage associated with at least one of the plurality of processing nodes.

The method includes providing a first message from the first processing node to other processing nodes at a first time. The first message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The replication counter in the first message indicates that, prior to the first time interval, all requests in the second request set associated with the second time interval from the plurality of state update request sets are replicated at two or more processing nodes. Any non-persistently stored requests associated with the time interval preceding the second time interval are replicated at two or more processing nodes. At least some requests in the first request set associated with the first time interval have not yet been replicated at two or more of these processing nodes.

The method includes providing a second message from the first processing node to other processing nodes at a second time following the first time interval. The second message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The replication counter value in the second message indicates that all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and any non-persistently stored requests associated with time intervals preceding the first time interval are replicated at two or more processing nodes. The second message causes at least one of the plurality of processing nodes to persistently store one or more requests in the first request set.

Each aspect may include one or more of the following characteristics.

The working counter can automatically increment the current time interval, and the replication counter can increment in response to messages received at the first processing node from other processing nodes among the plurality of processing nodes. For each time interval, each of the other processing nodes can maintain a first state update request count received at the processing node and a second state update request count sent from the processing node. The method may include: at the first processing node, receiving the first state update request count and the second state update request count for the first time interval from each of the other processing nodes; summarizing the received first state update request count and the second state update request count for the first time interval; and determining, based on the summarization, whether to increment the value of the replication counter.

The method may include incrementing the value of the replication counter from the value of the replication counter in the first message to the value of the replication counter in the second message. The method may include summing the received first state update request counts and second state update request counts within the first time interval, including calculating the difference between the sum of the received first state update request counts within the first time interval and the sum of the received second state update request counts within the first time interval. The method may include incrementing the value of the replication counter from the value of the replication counter in the first message to the value of the replication counter in the second message if the difference between the sum of the received first state update request counts within the first time interval and the sum of the received second state update request counts within the first time interval is zero.

Each of the other processing nodes can maintain an indicator of the latest time interval among the plurality of time intervals, wherein all state update requests received by that processing node and associated with the latest time interval have been persisted at that processing node. The method may include: at the first processing node, receiving the indicator of the latest time interval from each of the other processing nodes, and determining, based on the indicator of the latest time interval, whether to increment the persistence counter.

The method may include incrementing the persistence counter to the earliest time interval associated with the indicator of the latest time interval. The status update request may include one or more of the following: a data processing task, the result of the data processing task, and a data record.

In another general aspect, software for processing state update requests in a distributed data processing system comprising multiple processing nodes is stored in a non-transitory form on a computer-readable medium. The software includes instructions for causing the computing system to use two or more processing nodes to process multiple sets of requests. Each request in each set of requests is configured to cause a state update at one of the multiple processing nodes and is associated with a corresponding time interval among a plurality of time intervals. These sets of requests include a first set of requests associated with a first time interval among the plurality of time intervals.

The software also includes instructions for enabling the computing system to maintain multiple counters at a first processing node. These counters include: a working counter that indicates the current time interval among the multiple time intervals in the distributed data processing system; a replication counter that indicates a time interval among the multiple time intervals in which all requests associated with the time interval are replicated at several processing nodes among the multiple processing nodes; and a persistence counter that indicates a time interval among the multiple time intervals in which all requests associated with the time interval are stored in persistent storage associated with at least one of the multiple processing nodes.

The software also includes instructions for enabling the computing system to provide a first message from the first processing node to the other processing nodes among the plurality of processing nodes at a first time. The first message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The replication counter in the first message indicates that, prior to the first time interval, all requests in the second request set associated with the second time interval are replicated at two or more processing nodes, and any non-persistently stored requests associated with the time interval preceding the second time interval are replicated at two or more processing nodes. At least some of the requests in the first request set associated with the first time interval have not yet been replicated at two or more processing nodes among the plurality of processing nodes.

The software also includes instructions for causing the computing system to provide a second message from the first processing node to other processing nodes among the plurality of processing nodes at a second time following the first time interval. The second message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The value of the replication counter in the second message indicates that all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and any non-persistently stored requests associated with time intervals preceding the first time interval are replicated at two or more processing nodes. The second message causes at least one of the plurality of processing nodes to persistently store one or more requests in the first request set.

In another general aspect, an apparatus for processing state update requests in a distributed data processing system comprising multiple processing nodes includes the distributed data processing system comprising multiple processing nodes. The apparatus further includes one or more processors located at two or more of the multiple processing nodes for processing multiple sets of requests. Each request in each set of requests is configured to cause a state update at one of the multiple processing nodes and is associated with a corresponding time interval among a plurality of time intervals, the plurality of request sets including a first set of requests associated with a first time interval among the plurality of time intervals.

The apparatus further includes one or more data storage devices for maintaining multiple counters at a first processing node among the plurality of processing nodes. The multiple counters include: a working counter indicating the current time interval among the plurality of time intervals in the distributed data processing system; a replication counter indicating a time interval in which all requests associated with the time interval are replicated at several processing nodes among the plurality of processing nodes; and a persistence counter indicating a time interval in which all requests associated with the time interval are stored in a persistent storage device associated with at least one of the plurality of processing nodes.

The apparatus also includes a first output for providing a first message from the first processing node to the other processing nodes of the plurality of processing nodes at a first time. The first message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The replication counter in the first message indicates that, prior to the first time interval, all requests in the second request set associated with the second time interval are replicated at two or more processing nodes, and any non-persistently stored requests associated with the time interval preceding the second time interval are replicated at two or more processing nodes. At least some of the requests in the first request set associated with the first time interval have not yet been replicated at two or more processing nodes of the plurality of processing nodes.

The apparatus further includes a second output for providing a second message from the first processing node to other processing nodes among the plurality of processing nodes at a second time following the first time interval. The second message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The replication counter value in the second message indicates that all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and any non-persistently stored requests associated with time intervals preceding the first time interval are replicated at two or more processing nodes. The second message causes at least one of the plurality of processing nodes to persistently store one or more requests in the first request set.

In another general aspect, a computing system for processing state update requests in a distributed data processing system includes a plurality of processing nodes. The computing system includes means for using two or more of the plurality of processing nodes to process a plurality of request sets. Each request in each request set is configured to cause a state update at one of the plurality of processing nodes and is associated with a corresponding time interval among a plurality of time intervals. The plurality of request sets includes a first request set associated with a first time interval among the plurality of time intervals.

The computing system also includes means for maintaining a plurality of counters at a first processing node among the plurality of processing nodes. The plurality of counters includes: a working counter indicating the current time interval among the plurality of time intervals in the distributed data processing system; a replication counter indicating a time interval in which all requests associated with the time interval are replicated at several processing nodes among the plurality of processing nodes; and a persistence counter indicating a time interval in which all requests associated with the time interval are stored in persistent storage associated with at least one of the plurality of processing nodes.

The computing system also includes means for providing a first message from the first processing node to the other processing nodes of the plurality of processing nodes at a first time. The first message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The replication counter in the first message indicates that, prior to the first time interval, all requests in the second request set associated with the second time interval are replicated at two or more processing nodes, and any non-persistently stored requests associated with the time interval preceding the second time interval are replicated at two or more processing nodes. At least some of the requests in the first request set associated with the first time interval have not yet been replicated at two or more processing nodes of the plurality of processing nodes.

The computing system also includes means for providing a second message from the first processing node to other processing nodes among the plurality of processing nodes at a second time following the first time interval. The second message includes the value of the work counter, the value of the replication counter, and the value of the persistence counter. The value of the replication counter in the second message indicates that all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and that any non-persistently stored requests associated with the time interval preceding the first time interval are replicated at two or more processing nodes. The second message causes at least one of the plurality of processing nodes to persistently store one or more requests in the first request set.

Each aspect may have one or more of the following advantages.

Overall, compared to the methods described above where components (or parallel execution copies of components) are hosted on different servers, some features described herein enable improved computational efficiency (especially for computations based on graph-based program specifications). For example, a distributed data processing system comprising multiple processing nodes can increase the number of records processed per unit of given computational resources. For instance, a cluster component is deployed within a graph-based program specification and used to interface the graph-based program specification with the distributed data processing system, allowing the processing nodes within the graph-based program specification to perform the required computations in a distributed manner. Furthermore, some features described herein provide the ability to adapt to changing computational resources and requirements. This paper provides a computational approach that allows adaptation to changes in available computational resources during the execution of one or more graph-based computations, and/or to changes in computational load (e.g., due to the characteristics of the data being processed) or changes in the load of different components of such computations over time. For example, aspects can adapt to the addition or removal (or failure and re-online) of processing nodes from the distributed data processing system. One way to provide adaptability in a distributed data processing system is to manage the replication and persistence of data within the system, including maintaining a count of messages sent and received by the processing nodes and an indicator of the time interval at which all messages are replicated and/or persisted in the system.

A computational approach is also provided that efficiently utilizes computing resources with different characteristics (e.g., using servers with different numbers of processors per server, different numbers of processor cores per processor, etc.) and efficiently supports both homogeneous and heterogeneous environments. Some of the features described in this paper also enable rapid startup of graph-based computation. As described in this paper, one aspect of providing this efficiency and adaptability is the provision of appropriate management of the cluster of processing nodes.

Another advantage is its fault tolerance, as the distributed data processing system can recover from any processing errors that occur through timely rollback. The system anticipates multiple possible rollback scenarios and implements algorithms to perform a rollback in each of these scenarios.

Attached Figure Description

Figure 1 is a block diagram of a system used for data processing.

Figure 2 is a block diagram of a computing system that includes a computing cluster.

Figure 3 is a schematic diagram of a clock representing various repetition time intervals.

Figure 4 is a state transition diagram of the operation process.

Figures 5 to 12 illustrate the normal operation of the computing system.

Figures 13 to 15 illustrate the first rollback process.

Figures 16 to 18 illustrate the second rollback process.

Figures 19 to 21 illustrate the third rollback process.

Figures 22 to 25 illustrate the fourth rollback process.

Figures 26 to 29 illustrate the fifth rollback process.

Figures 30 to 32 illustrate the sixth rollback process.

Figures 33 to 35 illustrate the seventh rollback process.

Figures 36 and 37 illustrate the eighth rollback process.

Detailed Implementation

Figure 1 illustrates an example of a data processing system 200, in which computing cluster management technology can be used. System 200 includes a data source 202, which may include one or more data sources, such as storage devices or connections to online data streams. Each data source may store or provide data in any of a variety of formats, such as database tables, spreadsheet files, flat text files, or native formats used by the host. Execution environment 204 includes a preprocessing module 206 and an execution module 212. For example, execution environment 204 may be hosted on one or more general-purpose computers under the control of a suitable operating system, such as a version of the UNIX operating system. For example, execution environment 204 may include a multi-node parallel computing environment, which includes a configuration of a computer system using multiple processing units (e.g., central processing unit CPUs) or processor cores, which may be local (e.g., a multiprocessor system, such as a symmetric multiprocessing (SMP) computer), locally distributed (e.g., multiple processors coupled as a cluster or massively parallel processing (MPP) system), remote or remotely distributed (e.g., multiple processors coupled via a local area network (LAN) and/or a wide area network (WAN), or any combination thereof.

Preprocessing module 206 is capable of performing any configurations that may be required before execution module 212 executes the program specification (e.g., the graph-based program specification described below). Preprocessing module 206 can configure the program specification to receive data from various types of systems (including different forms of database systems) that can implement data source 202. Data can be organized as records with values (including possible null values) for corresponding fields (also called "attributes," "rows," or "columns"). When first configuring a computer program (such as a data processing application) to read data from a data source, preprocessing module 206 typically begins with some initial formatting information about the records in this data source. The computer program can be represented in the form of a data flow diagram as described herein. In some cases, the record structure of the data source may not be initially known but can be determined after analyzing the data source or data. Initial information about the records may include, for example, the number of bits representing different values, the order of fields within the record, and the type of value represented by bits (e.g., string, signed/unsigned integer).

The storage device providing the data source 202 may be local to the execution environment 204, for example, stored on a storage medium (e.g., hard disk 208) of a computer connected to the execution environment 204, or it may be remote from the execution environment 204, for example, hosted on a remote system (e.g., host 210) that communicates with the computer hosting the execution environment 204 via a remote connection (e.g., provided by cloud computing infrastructure).

Execution module 212 executes the program specifications configured and/or generated by preprocessing module 206 to read input data and/or generate output data. Output data 214 may be stored back in data source 202 or in a data storage system 216 accessible to execution environment 204, or otherwise used. Data storage system 216 may also be accessed by development environment 218, in which developer 220 can develop applications for using execution module 212 to process data.

In other words, the data processing system 200 may include:

An optional development environment 218 is coupled to the data storage system 216, wherein the development environment 218 is configured to build a data processing application that implements graph-based computation associated with a data flow graph. The graph-based computation is performed on data flowing from one or more input datasets through a graph processing graph component to one or more output datasets. The data flow graph is specified by a data structure in the data storage system 216, which has multiple nodes specified by the data structure and representing graph components connected by one or more links, which are specified by the data structure and represent data flows between graph components.

An execution environment 204 is coupled to a data storage system 216 and hosted on one or more computers. The execution environment 204 includes a preprocessing module 206 configured to read stored data structures of a specified data flow graph and to allocate and configure computational resources such as processes for performing computations on graph components of the data flow graph assigned by the preprocessing module 206.

The execution environment 204 includes an execution module 212, which schedules and controls the execution of assigned computations or processes to enable graph-based computations. Specifically, the execution module is configured to read data from the data source 202 and process the data using an executable computer program represented as a data flow graph.

1 Computing Cluster

Generally, some computer programs (also referred to herein as "applications") used to process data using execution module 212 include calling cluster components used by the application to access the computing cluster. For example, referring to FIG2, in a pipelined data processing method, calling cluster component 110 interacts with components of computer cluster 120 to process records 103 received at calling cluster component 110 from components in an application (e.g., a data flow graph or other form of graph-based program specification) to which calling cluster component 110 is a part, and transmits corresponding results 105 to one or more other components in the application to which calling cluster component 110 is a part. For each input record 103, calling cluster component 110 sends a request 113 to cluster 120 (e.g., a request to perform a data processing task), and after a period of time, the calling cluster component receives a response 115 from cluster 120 to the request 113. After a period of time following the receipt of response 115, typically after the result of the processing request is known to be properly persisted in cluster 120, calling cluster component 110 sends the result 105 corresponding to response 115.

Figure 2 does not show a graph-based programming specification in which the call cluster component 110 is a part. Only a single call cluster component 110 is shown in Figure 2; however, it should be recognized that there can typically be many call cluster components that can interact with the same cluster 120, for example, each participating in the same or different applications (such as data flow graphs). The graph-based programming specification can be implemented as, for example, a data flow graph as described in U.S. Patent Nos. 5,966,072, 7,167,850, or 7,716,630, or a data processing graph as described in U.S. Publication No. 2016/0062776. Such a data flow graph-based programming specification typically includes computational components corresponding to the nodes (vertices) of the graph, coupled via data flows corresponding to links (directed edges) in the graph (referred to as the “data flow graph”). Downstream components connected to upstream components via data flow links receive ordered streams of input data elements and process the input data elements in the received order to optionally generate one or more corresponding streams of output data elements. In some examples, each component is implemented as a process hosted on one of several typical computer servers. Each computer server can have multiple such component processes active at any given time, and the operating system (e.g., Unix) scheduler shares resources (e.g., processor time and/or processor cores) among the components hosted on that server. In this implementation, data flow between components can be implemented using data communication services (e.g., named pipes, TCP/IP sessions, etc.) of the operating system and the data network connecting the servers. A subset of components typically serves as a data source and/or data receiver for the entire computation (e.g., to and/or from data files, database tables, and external data streams). After component processes and data flow are established, for example, through cooperating processes, data flows through the entire computing system, which performs computations represented as graphs, typically governed by the availability of input data at each component and the scheduling of computational resources for each component.

Cluster 120 includes multiple cluster components 140, 150a to 150c coupled by a communication network 130 (shown as a "cloud" in Figure 2 and may have various interconnection topologies, such as bootstrapping, shared media, hypercubes, etc.). Each cluster component (or simply "component") has a specific role within the cluster. In some implementations, each component is hosted on a different computing resource (e.g., a single computer server, a single core of a multi-core server, etc.). It should be understood that these components represent roles within the cluster, and in some implementations, multiple roles may be hosted on a single computing resource, and a single role may be distributed across multiple computing resources.

In Figure 2, root component 140 (referred to as the “root”) performs some of the synchronization functions described in full below, but does not directly participate in the flow or computation of the data to be processed. Multiple worker components 150a to 150c (hereinafter referred to as “workers”) process requests 113 from the calling cluster component 110. Data 165 is stored redundantly in storage device 160 accessible to the respective worker 150, and each request 113 may require access (for reading and/or writing) to a specific portion of the data stored in storage device 160, identified by a key in the request 113, distributed across a specific subset of workers identified by the key. Among those workers holding the key data required for a particular request, one worker is designated as the primary worker (e.g., worker 150a) to execute request 113, while the others are designated as backup workers because these workers typically do not or need not execute the request, but their data versions are updated based on or in the same manner as the primary worker.

In Figure 2, the path of a specific input record 103 (which can be considered or includes the data unit to be processed) is shown as entering the call cluster component 110, then the component 110 sends the corresponding request 113 (along with the data unit) to the primary worker 150a (worker A) of the request, sends the response 115 from the primary worker 150a back to the call cluster component 110 and to the backup worker 150b (worker B) of the request, and finally outputs or sends the corresponding result 105 from the call cluster component 110. Typically, each request may have multiple backup components; however, for ease of illustration, only a single backup component is shown in many of the following examples.

As discussed further below, the invoking cluster component 110 caches request 113 in replay buffer 112 and, if necessary, can resend the request to cluster 120 to ensure that these requests have been correctly received and/or processed by cluster 120. Component 110 also caches response 115 in escrow buffer 114 and can receive redundant copies of certain responses in the event of an error state. Typically, component 110 holds responses "in escrow" until cluster 120 notifies component 110 that response 115 has been properly persisted in the cluster (i.e., stored in a data storage device with an appropriate durability level).

Root 140 performs synchronization by maintaining time (interval) values and assigning them to other components, as well as assigning certain time values to calling cluster component 110. Referring to Figure 3, clock 142 of root 140 maintains three times. Time T1 is the current working time or time interval, represented as an integer value, and is repeatedly updated, for example, incrementing once per second.

When cluster 120 receives request 113 from calling cluster component 110 and the cluster generates (or transmits) response 115, these requests and responses are each associated with the working time (T1) during which they were received and generated (or transmitted) (or equivalently with time intervals during which time T1 has the same value, i.e., between increments of T1). A second time T2 is maintained and allocated, which lags behind time T1. Time T2 represents a time (interval) such that all requests and/or responses created at or earlier times and sent between components 150a to 150c of cluster 120 have been replicated in multiple locations within components 150a to 150c (e.g., in volatile memory), so that these requests and responses do not need to be resent in the event of an operational rollback to handle errors, as described in more detail below. In some examples, replication (e.g., in volatile memory) is referred to as being stored in a data storage device at a first durability level. The root maintains and allocates a third time (interval) T3, which lags behind times T1 and T2. This time represents a period such that all requests and/or responses created at or earlier than this time have been stored in persistent memory and become permanent at at least one or all of the workers 150a to 150c of the data storage 165, so that these requests and responses do not need to be resent or recalculated in the event of an operational rollback for handling component failures in cluster 120. In some examples, being stored in persistent memory (e.g., stored to disk) is referred to as being stored in the data storage device at a second durability level, where the second durability level is relatively more durable than the first durability level. It should be noted that the data storage device may be associated with multiple different durability levels, which are relatively more or less durable than data storage devices with a first durability level and data storage devices with a second durability level. For example, a remote data storage device outside the cluster may have a third durability level, which is relatively more durable than the first and second durability levels. In some examples, time intervals T1, T2, and T3 can be alternatively referred to as "state consistency indicators".

The mechanism by which root 140 determines when to increase the replication time (T2) or persistence time (T3), and the mechanism by which the value of time (T1-T3) is assigned to workers 150a to 150c, are described later in this specification.

In normal operation, the request 113 received by cluster 120 is processed at worker 150, which is identified as the primary worker based on the key of the data unit of the request, and typically at one or more standby workers 150, also identified based on the key of the required data. Referring to Figure 4, this processing can be represented as transitions between different states of the request at the calling cluster component 110 and at the primary and standby workers 150. Note that different requests are in different states and are typically processed at different workers depending on the referenced data, and therefore, there may be many requests calling cluster components and any particular worker in different states.

Typically, each key is associated with a corresponding subset of workers 150 selected in a pseudo-random manner, for example, based on that key (e.g., a deterministic function of the key that assigns standby workers to each key value in an unpredictable manner). More generally and preferably, these subsets overlap with other subsets, rather than forming partitions of the complete set of workers based on key values.

When a request 113 with a unique identifier rid is formed at the invocation cluster component 110 for each input record 103, the request enters state A in the invocation cluster component. In the following description, each request 113 is in one of three states labeled A-C in the invocation cluster component, and at each of the workers 150 processing the request, it is in one of nine different states labeled A-I. After the invocation cluster component 110 records the request 113, it determines the worker 150 assigned as the primary worker for the request and sends the request 113 to that worker 150 (shown as worker A in Figure 2). Note that in an alternative embodiment, the invocation cluster component 110 may not know which worker is the designated primary worker, and the request 113 may be internally routed within the cluster 120 to reach the designated primary worker 150a. The request 113 remains in state A at the invocation cluster component 110 until a response 115 for the request is received from the cluster 120.

When request 113 is received at the primary worker (labeled worker A in Figure 2), the request enters state A at the primary worker. The primary worker assigns a request time, denoted as ta, to the request, which is equal to (it is known that) the current working time T1 allocated starting from root 140 (recognizing that there may be a time lag between the root increment T1 and the worker knowing that this increment). In this state, request 113 is stored in volatile memory 155 in association with request id rid, request time (denoted as ta in this example), and is designated as being in a state of waiting to be executed at the primary worker. In state A, the primary worker sends request 113 to one or more standby workers 150 (i.e., determined by a key) for the request. At the primary worker, resources are ultimately assigned to the request for execution, for example, based on the ordered allocation of resources according to the time (ta) assigned to the request and (optionally) the order in which the requests arrive at the primary worker. When request 113 begins execution on the primary worker, the request enters state B at the primary worker. When processing produces response 115, in this example, assuming the working time T1 is tb at this point, the requested state at the primary worker changes to state C. In state C, response 115 is stored in volatile memory 156 associated with time tb. As discussed further below, response 115 and any updates to the data storage device 160 at the worker are stored in association with time (here, time tb), in a manner that allows, for example, the use of a versioned database or other form of versioned data structure to eliminate the effects based on previous rollback times. In state C, response 115 is transmitted to the calling cluster component 110 and the standby components(s) 150.

At the calling cluster component 110, when a response 115 is received from the primary worker, the request enters state B, in which the response is stored in association with the time tb when the primary worker generated the response. Response 115 is held in the escrow buffer 114 at the calling cluster component until the calling cluster component receives an escrow time equal to or greater than tb from the root 140. Depending on the persistence requirements of the request from the calling cluster component, the root can provide a replication time T2 or a persistence time T3 as the escrow time for the calling cluster component. When the calling cluster component 110 receives an escrow time equal to or greater than tb, the calling cluster component sends out result 105, and the corresponding request 113 enters an empty state C, in which further recording of request 113 or its response 115 is no longer required (e.g., it can be completely deleted).

At standby worker(s) 150, when a standby worker receives a request 113 from the primary worker, the standby worker enters state F, in which the request is associated with the original request time ta (even if the increment of the current working time T1 exceeds this value), and the request is awaiting a response from the primary worker. When standby worker 150b receives a response 115 from the primary worker and thus copies the response 115 into its volatile memory 156, the request enters state G.

Once the primary or standby worker receives the newly generated response 115, it is free to begin the process of saving the response to persistent storage 160, such as a disk-based or non-volatile memory-based database or file system (see states D and H). A log-based approach can be used, where updates to persistent storage are first recorded in a volatile memory-based log, and portions of that log are periodically written to persistent storage 160. Note that even when a portion of the update log is written to persistent storage 160, those updates are not made permanent (i.e., "committed") until an explicit indicator of the degree to which an update is considered permanent is written to persistent storage.

Once root 140 has determined that all requests and responses associated with time tb and earlier have been replicated at all appropriate workers, T2 reaches or increases to tb. After allocating time T2 = tb from root 140 to the primary and standby workers 150, these workers permanently store the responses in persistent storage 160. If the update logs elapsed to time tb have not yet been written to persistent storage, they are written at that time. More generally, by the time T2 reaches or increases to tb, the logs elapsed to time tb have been written to persistent storage 160 by the workers, and all that must be done at this time is to make the updates permanent by recording an indicator that the updates elapsed to time tb in the persistent logs will be considered permanent. During the possible short period during which the primary worker is making the logs permanent, it is in state D. When the primary worker has stored the responses to the requests shown in Figure 4 in persistent storage, it enters state E. Similarly, when the standby worker makes the response permanent, it is in state H, and when the standby worker makes the response permanently stored in persistent memory, it enters state I. When the root determines that all responses associated with time tb (and earlier times) are permanently stored in persistent memory (i.e., all are in state E or I), the root increases the persistence time T3 to tb. As described above, for the case where the custody time of a request at the calling cluster component is persistence time T3, the root 140 notifies the calling cluster component 110 that the custody time is equal to or greater than tb, and the calling cluster component 110 releases the corresponding result 105 for the request 113 and response 115 to one or more other components within the application (e.g., the figure).

As described above, in normal operation, when processing consecutive requests 113 from calling cluster components in the cluster, the root update work time T1, the response 115 is returned to the calling cluster component, and released from the calling cluster component to the graph according to the update of the hosting time T2 or T3. Typically, processing a particular request 113 may take many time "ticks" of work time T1, such as tens or hundreds of ticks, and therefore the cluster may have many requests in progress, and many different request times associated with these requests. Furthermore, because the data is distributed among the workers, the load is effectively distributed among the workers according to the keys of those requests, such that each worker may have multiple requests that it is acting as the primary worker (i.e., in one of states A-E), and also multiple requests that it is acting as a backup worker (i.e., in one of states F-I).

It's important to note that some requests arriving at the cluster to execute tasks use procedures as described in this document to replicate the task and the corresponding results of executing it. For example, after a task has been tagged and replicated (but not necessarily persisted) at a standby worker, it is initialized at the primary worker. If the task operates on data records, initialization might involve retaining the original version 1 of the record. The task is then executed on the primary worker while remaining dormant on the standby worker. After processing is complete, a modified version 2 of the record will exist. Termination of the task can then include sending the modified version 2 of the record from the primary worker to the standby worker. Both the primary and standby workers are then able to delete the original version 1 of the record (and the replicated task). Each of these steps is reasonably efficient, but if the task duration is short, the overhead associated with these initialization and termination procedures can degrade task efficiency.

Alternatively, for some tasks with relatively short durations (“short tasks”), a different process can be used. Short tasks are still marked and copied at the standby worker. However, initialization does not require retaining the original version 1 of the record. Instead, after a commit operation indicates that the short task and its copy have been persistently stored on the primary and standby workers respectively, the short task is executed on both workers. At the end of this execution, a modified version 2 copy of the record will be available on both the primary and standby workers without any communication to transfer the modified record. Both workers have redundant processing, but since the task is short, this redundancy will not significantly impact efficiency. This alternative process is useful, for example, if the short task is deterministic and produces the same result regardless of which worker executes it.

2. Example of normal operation

Referring to Figures 5 through 12, an example of normal operation of the calling cluster component 110 and cluster 120 is shown. In Figure 5, input record 103 arrives at the calling cluster component 110, and the calling cluster component 110 forms a request 113 for input record 103. The calling cluster component 110 associates request 113 with a unique request identifier rid and stores it in the replay buffer 112 of the calling cluster component 110.

The calling cluster component 110 transmits request 113 to cluster 120, and receives the request at time T1 = ta on the primary worker 150a (worker A) in cluster 120. Request 113 is stored in the volatile memory 155 of the primary worker 150a and is assigned a request time equal to the current working time (T1 = ta). The request time of request 113 is provided to the calling cluster component 110, which associates the request time (i.e., ta) with request 113 stored in replay buffer 112. Request 113 stored in replay buffer 112 of the calling cluster component 110 is in state A (see Figure 4), awaiting a response from cluster 120. Request 113 stored in the volatile memory 155 of the primary worker is in state A, awaiting the allocation of computing resources to execute request 113.

Referring to Figure 6, the primary worker sends request 113 to the standby worker 150b (worker B), whereby the request is stored in the volatile memory 155 of the standby worker 150b. Request 113 stored in the volatile memory 155 of the standby worker 150b is in state F, awaiting a response from the primary worker.

Referring to Figure 7, once the primary worker 105 allocates computing resources (e.g., computing resources of the primary worker or another part of the cluster) to request 113, request 113 enters state B at the primary worker 105 and begins execution.

Referring to Figure 8, at time T1 = tb, the primary worker 105 completes the execution of request 113. The execution of request 113 generates a response 115, which is stored in the primary worker's volatile memory 156. Response 115 is associated with the request identifier (rid) of request 113 and the time (tb) at which the response was generated. The primary worker sends response 115 to the calling cluster component 110 and the standby worker 150b, and then request 113 is in state C, waiting for persistence time T3 to reach tb.

The calling cluster component 110 receives response 115 and stores it in its managed buffer 114. With the response stored in managed buffer 114, result 115 occurs when the calling cluster component 110 is in state B, waiting for the persistence time T3 (managed time in this example) to reach tb. The standby worker 150b receives response 115 and stores it in its volatile memory 156. Request 113 then enters state G at standby worker 150b, waiting for the persistence time T3 to reach tb.

Although not shown in Figure 8, in the case where the response 115 is stored (copied) in the volatile memory 156 of the primary worker 150a and the standby worker 150b, the copying time T2 is set to tb.

Referring to Figure 9, once the response 115 is stored in the volatile memory 156 of one or both of the primary worker 150a and the standby worker 150b, the primary worker 150a and the standby worker 150b begin to store the response 115 into their respective persistent storage devices 160, while also keeping it stored in their respective volatile memories 155, 156.

Referring to Figure 10, after storing response 115 at the primary worker and copying the response at the standby worker 150b, the persistence time T3 is set to tb. The primary worker 150a and the standby worker 150b ultimately permanently store response 115 in persistent storage 160. Request 113 stored at the primary worker is in state D, and request 113 stored at the standby worker 150b is in state H. In these states, request 113 and response 115 remain stored in volatile memories 155 and 156, respectively.

Referring to Figure 11, the escrow time in this example is a persistent time T3. Therefore, when T3 is updated to tb, the request 113 stored at the calling cluster component 110 enters state C, and the response 115 (which is associated with time tb) is released from the escrow buffer 114 of the calling cluster component.

Referring to Figure 12, when response 115 is permanently stored in the persistent storage of the primary worker 150a, request 113 enters state E, in which neither request 113 nor response 115 is stored in their respective volatile memories 155 and 156. Similarly, when response 115 is permanently stored in the persistent storage of the standby worker 150b, request 113 enters state I, in which neither request 113 nor response 115 is stored in their respective volatile memories 155 and 156.

3 Rollback Scenario

Although the state transition diagram in Figure 4 represents normal operation, it is possible (but rare) that messages between workers may not be successfully received. Furthermore, a worker may need to restart after losing its volatile memory, or the worker may fail completely, preventing it from processing further requests (i.e., assuming a primary or backup role). It should be noted that some embodiments of the data processing system described herein implement all the rollback scenarios described in this section. It should also be noted that other embodiments of the data processing system may implement one or more, but not all, of the rollback scenarios described in this section.

3.1 Scenario 1: tr < ta

First, consider the case where the cluster determines that some inter-worker messages were not successfully received and that these messages are associated with time ta. Generally, the root notifies all workers that they must "roll back" the time to a time tr prior to ta (i.e., tr < ta), for example, rolling back to tr = ta - 1. Even with this rollback, the results provided by the calling cluster component 110 can be provided to the application or graph as if no rollback had occurred, and updates to the data distributed among the workers remain consistent with the results provided by the calling cluster component. Specifically, the results are not released from the calling cluster component 110 to the application or graph until they are stored (e.g., replicated or persisted) across multiple nodes (e.g., workers), thus ensuring that the results will never be recalled or become invalid. In other words, any rollback that occurs must happen before the results are provided to the application or graph by the calling cluster component 110.

When root 140 determines that a rollback must be performed due to the failure to successfully receive some inter-worker messages, the root notifies the calling cluster component 110 of the rollback time tr. The current time T1 is incremented, and typically, all activity from time tr+1 up to and including T1-1 is considered as having not occurred. The effect at the calling cluster component 110 is that all requests in state B (i.e., the escrow time has not yet reached the response time) stored in replay buffer 112 are returned to state A, and any corresponding responses 115 in escrow buffer 114 are discarded. Requests 113 in state A (because these are already in state A or have returned from state B to state A) are then resent to cluster 120.

For the case where the request time *ta* is greater than the rollback time *tr* (i.e., *tr* < *ta*), we first consider the impact on requests that have not yet started execution but have been replicated between the primary and standby workers (i.e., the primary worker is in state A and the standby worker is in state F) within the cluster (i.e., at worker 150). In this illustration, the current working time is denoted as *tc*. Because *ta* is greater than *tr*, the calling cluster component cannot assume that the request has been correctly replicated and therefore removes the version of the request stored in the volatile memory 155 of the primary and standby workers. At cluster 120, a request 113 with the same request id *rid* is received from the calling cluster component 110 and associated with the new request time *tc*. When the primary worker receives request 113, it stores request 113 in its volatile memory 155 in state A. The primary worker sends request 113 to standby worker(s) 150, which stores request 113 in its volatile memory 155 in state F. Then, further processing is performed on the main and backup processors as shown in Figure 4.

Note that if the standby worker is unaware of the request before receiving the updated request with time tc from the primary worker, the standby worker will also proceed in the same manner as if the request had now been correctly replicated.

Referring to Figures 13 through 15, an example of a first rollback scenario is shown. In Figure 13, request 113, published at time ta, is stored in the replay buffer 112 at the calling cluster component 110 and is in state A. Request 113 is stored in the volatile memory 155 of the primary worker and is in state A because the request has not yet started execution. Request 113 is also stored in the standby worker 150b and is in state F.

A rollback request is received to roll back the system to time tr<ta. In Figure 14, after receiving the rollback request, request 113 is removed from the volatile memory 155 of the primary worker 150a and the volatile memory 155 of the standby worker 150b. The calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'. The new request 113' in the replay buffer 112 is in state A.

In the cluster, a new request 113' is sent to the primary worker. The primary worker 150a stores the new request 113' along with the request time tc in its volatile memory 155. The new request 113' stored in the volatile memory 155 of the primary worker 150a is in state A.

Referring to Figure 15, the primary worker sends a new request 113' to the standby worker 150b. The standby worker 150b stores the new request 113' in its volatile memory 155 and associates it with the request time tc. The updated request 113' stored in the standby worker's volatile memory 155 is in state F.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.2 Scenario 2: tr < ta, execution has started.

In the second scenario, the request time *ta* of the previous request is greater than the rollback time *tr* (i.e., *tr* < *ta*), but the request has already begun execution and has not yet completed on the primary worker (i.e., the request is in state B on the primary worker, where a partial response 115 may have been computed, while the request is in state F on the standby worker). In this case, execution is terminated on both the primary and standby workers, and the partial response 115 is discarded (or execution is allowed to complete, and the response is discarded), and cluster component 110 is invoked to resend request 113 to cluster 120. The requests stored on the primary and standby workers return to states A and F, respectively. The primary worker notifies the standby worker of the request in the same manner as if the request had not yet begun execution on the primary worker.

Referring to Figures 16 through 18, an example of a second rollback scenario is shown. In Figure 16, request 113, published at time ta, is stored in the replay buffer 112 at the calling cluster component 110 and is in state A. Request 113 is stored in volatile memory 155 at the primary worker 150a and is in state B because the request has begun execution. The request is also stored on the standby worker 150b and is in state F.

A rollback request is received to roll back the system to time tr<ta. In Figure 17, after receiving the rollback request, request 113 is removed from the volatile memory 155 of the primary worker 150a and the volatile memory 155 of the standby worker 150b. The calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'. The new request 113' in the replay buffer 112 is in state A.

In the cluster, a new request 113' is sent to the primary worker. The primary worker 150a stores the new request 113' along with the request time tc in its volatile memory 155. The new request 113' stored in the volatile memory 155 of the primary worker 150a is in state A.

Referring to Figure 18, the primary worker 150a sends a new request 113' to the standby worker 150b. The standby worker 150b stores the new request 113' in its volatile memory 155 and associates it with the request time tc. The updated request 113' stored in the standby worker's volatile memory 155 is in state F.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.3 Scenario 3: tr < ta < tb, execution completed

In the third case, the request time *ta* of the previous request is again greater than the rollback time *tr*. However, in this case, we assume that execution is completed at time *tb* (i.e., *tr* < *ta* ≤ *tb*), and the response has been replicated at the standby worker and received at the calling cluster component 110. That is, request 113 is in state B at the calling cluster component 110, in state C at the primary worker 150a, and in state G at the standby worker 150b. Instead of simply terminating the ongoing execution as in the second case, the response 115 stored at the primary and standby workers is removed. As described above with reference to Figure 4, the response generated at time *tb* is stored in a versioned data structure in association with time *tb* in a way that allows all updates at a specific time and later to be removed from the data structure. In this scenario, by removing all data versions updated at times later than time tr, the update to the requested data at time tb must also be removed. The request with time tc returns to state A at the primary worker, awaiting execution, and to state F at the standby worker, awaiting a response from the primary worker. At the point of invocation of the cluster component, the response is discarded, and the request returns to state A.

Referring to Figures 19 through 21, a simplified example of the third rollback scenario is shown. In Figure 19, request 113, published at time ta, is stored in replay buffer 112 at the calling cluster component 110. The response 115 generated at time tb is stored in managed buffer 114. Therefore, request 113 is in state B at the calling cluster component.

In the cluster, request 113 and response 115 are stored in volatile memories 155 and 156 at the primary worker 150a. Therefore, request 113 is in state C at the primary worker 150a. Request 113 and response 115 are also stored in volatile memories 155 and 156 at the standby worker. Therefore, the request is in state G at the standby worker 150b.

A rollback request is received to roll back the system to time tr < ta < tb. In Figure 20, after receiving the rollback request, response 115 is removed from the managed buffer 114 of the calling cluster component 110. In cluster 120, both request 113 and response 115 are removed from the volatile memory 155 of the primary worker 150a and the volatile memory 155 of the standby worker 150b.

The calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'. The new request 113' in replay buffer 112 is in state A.

In the cluster, a new request 113' is sent to the primary worker 150a. The primary worker 150a stores the new request 113' along with the request time tc in its volatile memory 155. The new request 113' stored in the volatile memory 155 of the primary worker 150a is in state A.

Referring to Figure 21, the primary worker 150a sends a new request 113' to the standby worker 150b. The standby worker 150b stores the new request 113' in its volatile memory 155 and associates it with the request time tc. The updated request 113' stored in the standby worker's volatile memory 155 is in state F.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.4 Scenario 4: ta < tr, execution has not yet started.

In the fourth scenario, the rollback time *tr* is at or after the original request time *ta* (i.e., *ta* ≤ *tr*), and the original request has not yet begun execution. The request is retransmitted to cluster 120, and is scheduled for execution after the original request (i.e., *{rid, ta}*) on both the primary and standby workers. The primary worker executes the original request and generates a response (i.e., *{rid, tb}*). Then, the primary worker proceeds to execute the retransmitted request (i.e., *{rid, tc}*), but detects that a response already exists associated with the *rid* of the retransmitted request and abandons the execution of the retransmitted request.

Referring to Figures 22 through 25, an example of a fourth rollback scenario is shown. In Figure 22, the original request 113, published at time ta, is stored in the replay buffer 112 at the calling cluster component 110 and is in state A. The original request 113 is also stored in the volatile memory 155 of the primary worker 150a and is in state A because the request has not yet started execution. The original request 113 is also stored in the standby worker 150b and is in state F.

A rollback request is received to roll back the system to time ta<tr. In Figure 23, the calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'. Request 113 in replay buffer 112 remains in state A.

In the cluster, a new request 113' is sent to the primary worker 150a. The primary worker 150a receives the new request 113' and schedules its execution after the original request 113. Both the original request 113 and the new request 113' are in state A in the volatile memory 155 of the primary worker 150a.

Referring to Figure 24, the primary worker 150a sends a new request 113' to the standby worker 150b. The standby worker 150b receives the new request 113' and schedules it for execution after the original request 113. Both the original request 113 and the new request 113' stored in the volatile memory 155 of the standby worker 150b are in state F.

Referring to Figure 25, the primary worker 150a has executed the original request 113 to generate a response 115, and the response 115 is persisted in the persistent storage 160 of the primary worker. As a result, the original request 113 is in state D on the primary worker 150a. The new request 113' has not yet begun execution on the primary worker 150a and is therefore in state A.

Response 115 has also been provided to standby worker 150b and calling cluster component 110. Standby worker 150b has stored response 115 in its volatile memory 156 and has persisted the response to its persistent storage device 160. Therefore, original request 113 is in state H on the standby worker. Calling cluster component 110 has stored response 115 in its managed buffer 114, and request 113' in the calling cluster component's replay buffer 112 is in state B.

When new request 113' begins execution at primary worker 150a, primary worker 150a recognizes that new request 113' is associated with the same request identifier rid as response 115, and therefore does not execute new request 113' because it is a copy. In some examples, response 115 can be retransmitted to the calling cluster component, which will ignore response 115 as a copy.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.5 Scenario 5: ta < tr, execution has started.

In the fifth scenario, the rollback time tr is at or after the original request time ta (i.e., ta ≤ tr), and the original request has begun execution but has not yet completed at the primary worker (i.e., the request is in state B at the primary worker and in state F at the standby worker). In this case, execution is terminated (or allowed to complete and the response is discarded) at both the primary and standby workers (i.e., the requests stored at the primary and standby workers are returned to states A and F, respectively).

The request is retransmitted to cluster 120 via cluster component 110, where it is executed after the original request (i.e., {rid, ta}) at both the primary and standby workers. The primary worker executes the original request and generates a response (i.e., {rid, tb}). Then, the primary worker proceeds to the request to begin retransmission (i.e., {rid, tc}), but detects that a response already exists associated with the rid of the retransmitted request and abandons the retransmission.

Referring to Figures 26 through 29, an example of a fifth rollback scenario is shown. In Figure 26, the original request 113, published at time ta, is stored in the replay buffer 112 at the calling cluster component 110 and is in state A. The original request 113 is stored in volatile memory 155 at the primary worker 150a and is in state B because the request has begun execution. The original request 113 is also stored in the standby worker 150b and is in state F.

A rollback request is received to roll back the system to time ta<tr. In Figure 27, the calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'. Request 113 in replay buffer 112 remains in state A.

In cluster 120, the execution of the original request 113, stored in the volatile memory 155 of the primary worker 150a, is terminated, and the original request 113 returns to state A. A new request 113' is sent to the primary worker 150a. The primary worker 150a receives the new request 113' and schedules it for execution after the original request 113. The new request 113' stored in the volatile memory 155 of the primary worker 150a is in state A.

Referring to Figure 28, the primary worker 150a sends a new request 113' to the standby worker 150b. The standby worker 150b receives the new request 113' and schedules it for execution after the original request 113. Both the original request 113 and the new request 113' stored in the volatile memory 155 of the standby worker 150b are in state F.

Referring to Figure 29, the primary worker 150a has executed the original request 113 and generated a response 115. The response 115 is persisted in the persistent storage 160 of the primary worker. As a result, the original request 113 is in state D on the primary worker 150a. The new request 113' has not yet begun execution on the primary worker 150a and is therefore in state A.

The response 115 has also been copied to the standby worker 150b and the calling cluster component 110. The standby worker 150b has stored the response 115 in its volatile memory 156 and has persisted the response to its persistent storage device 160. Therefore, the original request 113 is in state H on the standby worker. The calling cluster component 110 has stored the response 115 in its managed buffer 114, and request 113' is in state B in the calling cluster component's replay buffer 112.

When new request 113' begins execution at primary worker 150a, primary worker 150a recognizes that new request 113' is associated with the same request identifier rid as response 115, and therefore does not execute new request 113' because it is a copy. In some examples, response 115 may be retransmitted to cluster component 110, which will ignore response 115 as a copy.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.6 Scenario 6: ta < tb < tr, execution completed

In the sixth scenario, the rollback time *tr* is at or after the request time *ta*, and the request has been completed at time *tb*, which is also at or before the rollback time (i.e., *ta* ≤ *tb* ≤ *tr*). If the response has been successfully provided to the calling cluster component 110 (i.e., the request was in state B when the calling cluster component was in state B), the rollback request will not cause the request to be resent, nor will it cause any response to be removed from the managed buffer 114. That is, any requests associated with *ta* and any responses associated with *tb* remain unchanged.

However, if the response is not successfully provided to the calling cluster component 110, the calling cluster component 110 retransmits the request to cluster 120. When the primary worker receives the retransmitted request, it begins to execute the retransmitted request (i.e., {rid, tc}), but detects that a response 115 already exists associated with the request identifier rid. Therefore, the retransmitted request is not executed, and the response generated by executing the original request is retransmitted to the calling cluster component 110. The calling cluster component 110 receives the response with a response time tb, which is used to determine when a response can be sent from the managed service at the calling cluster component.

Referring to Figures 30 to 32, an example of a sixth rollback scenario is shown. In Figure 30, the original request 113, published at time ta, is stored in the replay buffer 112 at the calling cluster component 110. A response 115 to the original request 113 is generated at time tb, but this response does not reach the managed buffer 114 of the calling cluster component 110. Therefore, request 113 is in state A at the calling cluster component 110.

In the cluster, request 113 and response 115 are stored in volatile memories 155 and 156 of the primary worker 150a. Therefore, request 113 is in state C on the primary worker 150a. Request 113 and response 115 are also stored in volatile memories 155 and 156 at the standby worker. Therefore, the request is in state G on the standby worker 150b.

A rollback request is received to roll back the system to time ta < tb < tr. In Figure 31, the calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'.

A new request 113' is sent to the primary worker 150a in cluster 120. The primary worker 150a receives the new request 113' and queues it in volatile memory 155 for execution. The original request 113 stored in the volatile memory 155 of the primary worker 150a remains in state C, and the new request 113' stored in the volatile memory 155 of the primary worker 150a is in state A.

Referring to Figure 32, when the primary worker 150a begins executing a new request, it recognizes that the new request 113' has the same request identifier rid as the original request 113, and that a response 115 associated with the request identifier rid already exists at the primary worker 150a. Therefore, the primary worker 150a does not execute the new request 113', but instead retransmits the response 115 to the call cluster component 110. The call cluster component 110 receives the response 115 and stores it in a managed buffer 114. With the response 115 stored in the managed buffer 114 of the call cluster component 110, the call cluster component 110 is in state B.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.7 Scenario 7: ta < tr < tb, execution completed

In the seventh scenario, the rollback time tr is at or after the request time ta, and the request has already been executed at a time tb after the rollback time (i.e., ta ≤ tr < tb). The replication of the response between workers may not have succeeded. The workers discard all responses 115 with a time after tr. Request 113 stored in the standby worker returns to state F, and request 113 stored in the primary worker returns to state B. Cluster component 110 discards all responses 115 in the managed buffer 114, returns request 113 stored in the replay buffer 112 to state A, and resends request 113 to cluster 120, which reprocesses the request.

Referring to Figures 33 to 35, an example of the seventh rollback scenario is shown. In Figure 33, request 113, published at time ta, is stored in replay buffer 112 at the calling cluster component 110. The response 115 generated at time tb is stored in managed buffer 114. Therefore, request 113 is in state B at the calling cluster component 110.

In cluster 120, request 113 and response 115 are stored in volatile memories 155 and 156 at the primary worker 150a. Therefore, request 113 is in state C at the primary worker 150a. Request 113 is also stored in volatile memories 155 and 156 at the standby worker 105, but response 115 may have been or may not have been successfully copied to the standby worker 150b. Therefore, the request may or may not be in state G at the standby worker 150b.

A rollback request is received to roll back the system to time ta < tr < tb. In Figure 34, the response 115 stored in the managed buffer 114 of the calling cluster component 110 is removed. The calling cluster component 110 publishes a new request 113' associated with the same request identifier (rid) as the original request 113 to cluster 120. At time tc, the new request 113' is received by cluster 120 and associated with the request time tc. Cluster 120 notifies the calling cluster component 110 of the request time tc associated with the new request 113'. The new request 113' in the replay buffer 112 is in state A.

In cluster 120, standby worker 150b removes any responses associated with the time following tr from its volatile memory 156 and thus returns to state F. Primary worker 150a returns to state B. A new request 113' is sent to primary worker 150a. Primary worker receives new request 113' and schedules it for execution after the original request 113. The new request 113' stored in volatile memory 155 of primary worker 150a is in state A.

In Figure 35, the primary worker 150a completes the execution of the original request 113 at time td and generates a new response 115'. The primary worker 150a sends the new response 115' to the standby worker 150b and invokes the cluster component 110 to transition the state of the original request 113 stored in the volatile memory of the primary worker 150a to state C. The standby worker 150b receives the new response 115' and stores it in its volatile memory 155 to transition the state of the original request 113 stored in the standby worker's volatile memory 155 to state G. The invoked cluster component 110 receives the new response 115' and stores it in the managed buffer 114 to transition the state of the new request 113' stored in the replay buffer 112 to state B.

When the new request 113' begins execution at the main worker 150a, the main worker 150a recognizes that the new request 113' has the same request identifier rid as the original request 113, and therefore does not execute the new request 113' because it is a copy.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

3.8 Scenario 8: ta < tr < tb, execution completed

Finally, in the eighth case, the worker processing the request as the primary worker is lost (e.g., known failure). Generally, for any request at a standby worker that is waiting for the lost primary worker to provide a response (i.e., the standby worker is in state F), that standby worker is promoted to primary worker. When root 140 detects a worker loss, for example due to failure to receive a response to a message from that worker, the root initiates a rollback to time tr, which is equal to the time of the last replication (i.e., tr = T2). When the standby worker receives a rollback request to time tr (possibly accompanied by new partition information to accommodate the lost worker), the standby worker begins to act as the new primary worker by changing the state of the request to state A, where it is waiting for resources to execute the request.

Referring to Figures 36 and 37, an example of an eighth rollback scenario is shown. In Figure 36, request 113, issued at time ta, is stored in replay buffer 112 at the calling cluster component 110 and is in state A. Request 113 is stored in volatile memory 155 at the primary worker 150a and is in state B because the request has started execution but has not yet completed. The request is also stored in standby worker 150b and is in state F. During the execution of request 113, the primary worker 150a fails or is lost.

In Figure 37, the root has requested a rollback to time tr, which is equal to the time of the last replication. At this point, standby worker 150b is promoted to primary worker 150a and its state is changed to state A. Another worker 150c is assigned as a standby worker in state F.

Then, the cluster proceeds according to its normal operation (as illustrated in Figures 5 to 12).

4 nodes

Now, turning to the operation of root 140, as described above, the root periodically increments the current working time (interval) T1 144. Generally, when the root updates its working time, it distributes (e.g., broadcasts) time tuples (T1, T2, T3) 144-146 to all workers. In response, the workers provide information to the root, which can then update the T2 and/or T3 times based on that information.

Each worker maintains a set of counters 151-152 associated with a specific work time. One counter 151, associated with work time t1 (called Sent(t1)), counts the number of communications sent from that worker to the standby worker with a request time t1, and the number of responses sent to the standby worker with a response time t1. In Figure 4, under state A, Sent(ta) is updated for each request sent to the standby worker with a request time ta, and Sent(tb) is incremented for each response generated at time tb and sent for replication at the standby worker. Note that the Sent() counter is not incremented for messages sent from the worker to invoke cluster components. Another counter 152, Rec(t1), counts the number of communications received at the worker associated with time t1. Specifically, when a standby worker receives a copy of a request with a requested time ta when it enters state F, the standby worker increments Rec(ta), and when a standby worker receives a copy of a response generated at time tb when it enters state G, the standby worker increments Rec(tb). Each worker has its own local copy of these counters, denoted as Sentw(t) and Recw(t) for worker w. It should be apparent that, to the extent that all communications sent in association with time t1 are also received at their destination, the sum of Sentw(t) for all workers w is equal to the sum of Recw(t) for worker w.

Occasionally, for example, in response to a broadcast of the current time (T1, T2, T3) received from root 140, each worker 150 sends its current counts Sent(t) 151 and Rec(t) 152 for all times greater than the replication time T2. These counts are received and summed at the root, such that the root determines the sum of Sent(t) and the sum of Rec(t) for each time t greater than T2 and stores them in counters 141 and 142 in association with the corresponding time. If Sent(T2+1) equals Rec(T2+1), then all transmissions from time T2+1 have been received, and T2 is incremented to become the next replication time. This process is repeated until Sent(T2+1) is not equal to Rec(T2+1) or T2+1 reaches T1. This incremented T2 time (145) is then used for the next broadcast starting from the root.

As described above, data updates at the worker's location are first recorded in volatile memory, with logs periodically written to persistent storage. For changes prior to replication time T2, each worker is free to persist the recorded changes in persistent storage. Typically, each worker w has already had the opportunity to persist all changes up to time T3(w), where different workers typically reach different times. In addition to returning Rec() and Sent() to the root in response to the broadcast of the current time, each worker also returns its T3(w) time, which is summed based on the min() operation at the root or on the communication path back to the root. That is, the root determines T3 = minw T3(w), and then the root assigns this new value of T3 the next time the current time is allocated.

In some embodiments, the root distributes time tuples (T1, T2, T3) in direct (e.g., unicast) communication between the root and each worker. In other embodiments, tuples are distributed in a different manner (such as flood-based broadcasting). In yet another embodiment, tuples are distributed along a predetermined tree-structured distribution network, where each receiver of a tuple forwards it to multiple other receivers, such that eventually all workers have received the time tuples.

The summation of counts from the workers can be performed via unicast communication between each worker and the root node, where the root performs a full summation over all workers. As a more efficient solution, the counts can be sent back along the same path as the time tuple, where intermediate nodes in the path perform partial summations of the counts, thus sharing the burden of summation with the root, but still obtaining a summation of counts from all workers.

In alternative operating modes, responses can be released from the calling cluster component when replicating response time rather than persistence time. In this way, responses can be provided to the graph with minimal latency, even if the response may not yet be persisted in cluster storage.

As described above, the response to the execution request is stored in a versioned data structure. In such a data structure, each update to a data item is stored as a separately recoverable version, and that version is tagged with the time associated with that update. For example, the data structure can be stored, at least conceptually, as a list of tuples (tb values) for each access key, where tb is the update time of the value. Values at different times can share substructures or use other storage optimizations. In some examples, these values are stored based on edits to the data values between times. As an example, these values can be represented as a tree-based structure, and each version can be stored as a "forward" incremental operation sufficient to create the next version from a previous version, or as a "backward" incremental operation sufficient to reconstruct a previous version from the current version. As discussed above, this versioned data structure allows all updates to be rolled back after a rollback time. Instead of maintaining all updates to the data item, only updates relative to the beginning of the update time are retained so that a rollback to the beginning of any update time can be completed.

It should be recognized that after adding a replication time T2 to the root, the worker will not be required to roll back to that time or a version prior to it. Therefore, an optimization of the versioned data structure is to remove versions with replication time T2 or earlier from the data structure.

In some embodiments, certain requests are "lightweight" in the sense that their execution time is small, and therefore, executing the request at a standby worker can consume fewer resources than replicating the response from the primary worker to the standby worker. In this embodiment, response replication from the primary worker to the standby worker(s) is not performed. Each worker may complete processing at a different time. To maintain data synchronization between workers, the primary worker allocates a completion time tb as described above, while the standby workers treat their locally computed responses as if those responses were computed at that time.

In an alternative embodiment, the invoking cluster component participates in the cluster in the sense that it receives time tuples from the root and returns Sent() and Rec() counts to the root. In this embodiment, the invoking cluster component dispatches request times for requests, which are used by the workers during request replication. When a rollback occurs, because the invoking cluster component knows the request times of the requests it holds, it only needs to resend the request after the rollback time, without discarding responses generated at or before the rollback time. The worker operation is modified to accommodate this operation of the invoking cluster component.

5 alternative solutions

More generally, in the rollback scenarios 4-8 above, where ta < tr, when the calling cluster component 110 retransmits the request, the calling cluster component is unaware (and indifferent) that the original request was transmitted at time ta. On the other hand, cluster 120 needs to consider the request time of the original request, as the cluster uses this time to determine whether to rollback. Therefore, when the calling cluster component 110 retransmits the request (with the request identifier rid) to cluster 120 such that ta < tr < tc, the request is received at the primary worker 150a and associated with time tc. The primary worker 150a forwards the request to the standby worker 150b. In this case, the primary worker can execute the original request (i.e., {rid, ta}) before executing the retransmitted request (i.e., {rid, tc}). When the primary worker 150a proceeds to execute the retransmitted request (i.e., {rid, tc}), because the response to the original request (i.e., {rid, ta}) has been persisted, the primary worker treats the retransmitted request as a copy.

In some examples, the request spawns subsequent tasks (sometimes called a 'task chain'). In such examples, a response to the request is not generated until after the spawned tasks have completed. In some examples, if a response to the request {rid,ta} has already been stored, the worker returns its response to the calling cluster component. However, if a response to the request {rid,ta} does not yet exist because the request {rid,ta} has not yet completed, the subsequent request {rid,tc} with a copy of rid is ignored because the cluster knows that the original request will eventually complete and generate a response to be returned to the calling cluster component.

In the example above, when the cluster receives a request, it associates a time (e.g., ta) with the request and then notifies the calling cluster component of that time. The calling cluster component associates the time with the request stored in its replay buffer. In the case of a rollback, the calling cluster component can selectively replay the request using the time associated with the request in its replay buffer. However, in some examples, neither the cluster nor the calling cluster component associates the request with a time. In these examples, the calling cluster component has less selectivity when replaying requests in a rollback scenario. For example, in the case of a rollback request, the calling cluster component could systematically replay all requests in its replay buffer.

6 Implementation Methods

The aforementioned computing cluster management method can be implemented, for example, using a programmable computing system that executes suitable software instructions, or the method can be implemented in suitable hardware, such as a field-programmable gate array (FPGA) or in some hybrid form. For example, in a programmable method, the software may include processes in one or more computer programs that execute on one or more programmable or controlled computing systems (which can be of various architectures, such as distributed client/server systems or power grids), each 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 may include one or more modules of a larger program that, for example, provides services related to the design, configuration, and execution of a data flow graph. Program modules (e.g., elements of a data flow graph) may be implemented as data structures or other organized data conforming to a data model stored in a data repository.

Software can be stored in a non-transitory form for a period of time (e.g., the time between refresh cycles of a dynamic memory device such as dynamic RAM), such as in volatile or non-volatile storage media embodied by the physical properties of the medium (e.g., surface pits and shores, magnetic domains, or charges), or any other non-transitory medium. When preparing to load instructions, the software can be provided on a tangible, non-transitory medium such as a CD-ROM or other computer-readable medium (e.g., readable by a general-purpose or special-purpose computing system or device), or delivered via a communication medium over a network (e.g., encoded into a propagating signal) to the tangible, non-transitory medium of the computing system to which it is executed. Some or all of the processing can be executed on a dedicated computer or using dedicated hardware such as a coprocessor or field-programmable gate array (FPGA) or dedicated application-specific integrated circuit (ASIC). The processing can be implemented in a distributed manner, where different parts of the computation specified by the software are executed by different computing elements. Each such computer program is preferably stored or downloaded to a computer-readable storage medium (e.g., solid-state memory or medium, or magnetic or optical medium) accessible by a general-purpose or special-purpose programmable computer, so as to configure and operate the computer when the storage medium is read by the computer to perform the processes described herein. The system of the present invention may also be considered as a tangible, non-transitory medium configured with a computer program, wherein such a medium causes the computer to operate in a specified and predefined manner to perform one or more of the processing steps described herein.

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

Claims (14)

1. A method for processing state update requests in a distributed data processing system comprising multiple processing nodes, the method comprising: Multiple request sets are processed using two or more of the multiple processing nodes, each request in each request set is configured to cause a state update at one of the multiple processing nodes and associated with a corresponding time interval among multiple time intervals, the multiple state update request sets including a first request set associated with a first time interval among the multiple time intervals; Multiple counters are maintained at the first processing node among the plurality of processing nodes, and the plurality of counters include: A working counter, which in the distributed data processing system indicates the current time interval and its value among a plurality of time intervals. A replication counter, indicating a specific time interval among a plurality of time intervals and the value of that time interval, during which all requests associated with that time interval are replicated at several processing nodes among the plurality of processing nodes, and A persistence counter, indicating a specific time interval among the plurality of time intervals, during which all requests associated with that time interval are stored in persistent storage associated with at least one of the plurality of processing nodes. A first message is provided from the first processing node to the other processing nodes among the plurality of processing nodes at a first time. The first message includes a first value of the working counter, a first value of the replication counter, and a first value of the persistence counter at the first time, wherein the first value of the replication counter lags behind the first value of the working counter and the first value of the persistence counter lags behind the first value of the replication counter, and wherein the replication counter in the first message indicates: Prior to the first time interval, in response to the determination that a first value of the replication counter is equal to or greater than an earlier time interval associated with the second request set, all requests in the second request set associated with the earlier time interval from the plurality of state update request sets are replicated at two or more processing nodes. Replicate any non-persistent storage requests associated with a time interval preceding the earlier of the multiple time intervals at two or more processing nodes. Where at least some requests in the first request set associated with the first time interval have not yet been replicated at two or more of the plurality of processing nodes, and A second message is provided from the first processing node to the other processing nodes among the plurality of processing nodes at a second time following the first time. The second message includes a second value of the work counter, a second value of the replication counter, and a second value of the persistence counter at the second time, wherein the second value of the replication counter lags behind the second value of the work counter and the second value of the persistence counter lags behind the second value of the replication counter. The value of the replication counter in the second message indicates: In response to the determination that a second value of the replication counter is equal to or greater than the first time interval associated with the first request set, all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and Replicate any non-persistent storage requests associated with time intervals preceding the first time interval at two or more processing nodes. In response to the determination that a second value of the persistence counter is equal to or greater than the first time interval associated with the first request set, the second message causes at least one of the plurality of processing nodes to complete a storage operation to persistent storage for one or more requests in the first request set. 2. The method of claim 1, wherein the working counter automatically increments the current time interval, and the replication counter increments in response to a message received at the first processing node from other processing nodes among the plurality of processing nodes. 3. The method as claimed in claim 1 or 2, wherein, for each of the plurality of time intervals, each of the other processing nodes maintains a count of first state update requests received at the processing node and a count of second state update requests sent from the processing node. 4. The method as described in claim 1 or 2, further comprising: at the first processing node, receiving a first state update request count and a second state update request count within the first time interval from each of the other processing nodes; summarizing the received first state update request count and second state update request count within the first time interval; and determining, based on the summarization, whether to increment the value of the replication counter. 5. The method of claim 1 or 2, further comprising: incrementing the value of the replication counter from the value of the replication counter in the first message to the value of the replication counter in the second message. 6. The method of claim 4, wherein summing the received first state update request counts and second state update request counts within the first time interval includes calculating the difference between the sum of the received first state update request counts within the first time interval and the sum of the received second state update request counts within the first time interval. 7. The method of claim 6, further comprising: if the difference between the sum of the first state update request counts received within the first time interval and the sum of the second state update request counts received within the first time interval is zero, then incrementing the value of the replication counter from the value of the replication counter in the first message to the value of the replication counter in the second message. 8. The method of claim 1 or 2, wherein each of the other processing nodes maintains an indicator of the latest time interval among the plurality of time intervals, wherein all state update requests received by the processing node and associated with the latest time interval have been persisted at the processing node. 9. The method of claim 8, further comprising: at the first processing node, receiving an indicator of the latest time interval from each of the other processing nodes, and determining whether to increment the persistence counter based on the indicator of the latest time interval. 10. The method of claim 9, further comprising: incrementing the persistence counter to the earliest time interval associated with the indicator of the latest time interval. 11. The method of claim 1 or 2, wherein the status update request includes one or more of a data processing task, a data processing task result, and a data record. 12. A computer-readable medium having stored thereon computer-executable instructions for processing state update requests in a distributed data processing system comprising multiple processing nodes, the computer-executable instructions, when executed by a computer, causing the computer to perform the steps of the method according to any one of claims 1 to 11. 13. An apparatus for processing state update requests in a distributed data processing system comprising multiple processing nodes, the apparatus comprising: This distributed data processing system includes the multiple processing nodes; One or more processors, the one or more processors being located at two or more of the plurality of processing nodes and being configured to process a plurality of request sets, each request in each request set being configured to cause a state update at one of the plurality of processing nodes and being associated with a corresponding time interval of a plurality of time intervals, the plurality of request sets including a first request set associated with a first time interval of the plurality of time intervals; One or more data storage devices are configured to maintain a plurality of counters at a first processing node among the plurality of processing nodes, the plurality of counters including: A working counter, which in the distributed data processing system indicates the current time interval and its value among a plurality of time intervals. A replication counter, indicating a specific time interval among a plurality of time intervals and the value of that time interval, during which all requests associated with that time interval are replicated at several processing nodes among the plurality of processing nodes, and A persistent counter, indicating a specific time interval among the plurality of time intervals and the value of that time interval, during which all requests associated with that time interval are stored in persistent storage associated with at least one of the plurality of processing nodes. A first output, configured to provide a first message from the first processing node to other processing nodes among the plurality of processing nodes at a first time, the first message including a first value of the working counter at the first time, a first value of the replication counter at the first time, and a first value of the persistence counter at the first time, wherein the first value of the replication counter lags behind the first value of the working counter and the first value of the persistence counter lags behind the first value of the replication counter, and wherein the replication counter in the first message indicates: Prior to the first time interval, in response to the determination that a first value of the replication counter is equal to or greater than an earlier time interval associated with the second request set, all requests in the second request set associated with the earlier time interval are replicated at two or more processing nodes. Replicate any non-persistent storage requests associated with a time interval preceding the earlier of the multiple time intervals at two or more processing nodes. Where at least some requests in the first request set associated with the first time interval have not yet been replicated at two or more of the plurality of processing nodes, and A second output is provided from the first processing node to other processing nodes among the plurality of processing nodes at a second time after the first time. The second message includes a second value of the working counter, a second value of the replication counter, and a second value of the persistence counter at the second time, wherein the second value of the replication counter lags behind the second value of the working counter and the second value of the persistence counter lags behind the second value of the replication counter. The value of the replication counter in the second message indicates: In response to the determination that a second value of the replication counter is equal to or greater than the first time interval associated with the first request set, all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and Replicate any non-persistent storage requests associated with time intervals preceding the first time interval at two or more processing nodes. In response to the determination that a second value of the persistence counter is equal to or greater than the first time interval associated with the first request set, the second message causes at least one of the plurality of processing nodes to complete a storage operation to persistent storage for one or more requests in the first request set. 14. A computing system for processing state update requests in a distributed data processing system comprising multiple processing nodes, the computing system comprising: A means for using two or more of the plurality of processing nodes to process a plurality of request sets, each request in each request set being configured to cause a state update at one of the plurality of processing nodes and associated with a corresponding time interval of a plurality of time intervals, the plurality of request sets including a first request set associated with a first time interval of the plurality of time intervals; A means for maintaining a plurality of counters at a first processing node among the plurality of processing nodes, the plurality of counters including: A working counter, which in the distributed data processing system indicates the current time interval and its value among a plurality of time intervals. A replication counter, indicating a specific time interval among a plurality of time intervals and the value of that time interval, during which all requests associated with that time interval are replicated at several processing nodes among the plurality of processing nodes, and A persistent counter, indicating a specific time interval among the plurality of time intervals and the value of that time interval, during which all requests associated with that time interval are stored in persistent storage associated with at least one of the plurality of processing nodes. A means for providing a first message from a first processing node to other processing nodes among a plurality of processing nodes at a first time, the first message including a first value of a working counter at the first time, a first value of a replication counter at the first time, and a first value of a persistence counter at the first time, wherein the first value of the replication counter lags behind the first value of the working counter and the first value of the persistence counter lags behind the first value of the replication counter, and wherein the replication counter in the first message indicates: Prior to the first time interval, in response to the determination that a first value of the replication counter is equal to or greater than an earlier time interval associated with the second request set, all requests in the second request set associated with the earlier time interval are replicated at two or more processing nodes. Replicate any non-persistent storage requests associated with a time interval preceding the earlier of the multiple time intervals at two or more processing nodes. Where at least some requests in the first request set associated with the first time interval have not yet been replicated at two or more of the plurality of processing nodes, and A means for providing a second message from the first processing node to other processing nodes among the plurality of processing nodes at a second time following the first time, the second message including a second value of the working counter at the second time, a second value of the replication counter at the second time, and a second value of the persistence counter at the second time, wherein the second value of the replication counter lags behind the second value of the working counter and the second value of the persistence counter lags behind the second value of the replication counter, and wherein the value of the replication counter in the second message indicates: In response to the determination that a second value of the replication counter is equal to or greater than the first time interval associated with the first request set, all requests in the first request set associated with the first time interval are replicated at two or more processing nodes, and Replicate any non-persistent storage requests associated with time intervals preceding the first time interval at two or more processing nodes. Wherein, if the second value of the persistence counter is determined to be equal to or greater than the first time interval associated with the first request set, the second message causes at least one of the plurality of processing nodes to complete a storage operation to persistent storage device for one or more requests in the first request set.