Patent Description:
This specification relates to distributed computing systems, data processing, and event ordering.

A difficult problem in distributed computing and multi-agent systems is to maintain the causal relationship between events that occur within the systems. This issue can be present due to the limited or absence of access to a universal physical clock that is used to maintain time throughout the system and/or to synchronize clocks throughout the system.

<CIT> discloses a distributed event processing system that can organize (e.g., order) input streams regardless of actual time of receipt. This order may simply be arrival order or given explicitly on a specific event attribute, such as a timestamp or sequence number. The disclosure employs punctuation and heartbeats in connection with an event processing system. The disclosure discloses mechanisms by which heartbeats and timestamps can be regularly generated by low-level nodes (e.g., sources) and propagated through the network, to unblock standing event pattern queries and align events from multiple distributed streams.

This specification generally describes systems and techniques for ordering events using a disentanglement protocol. This allows distributed systems and high-level nodes to infer the order and/or time of occurrence of events from the perspective of the high-level nodes. The approaches described in this document use causal event relationships, which alleviates the problem of non-synchronized clocks and the lack of access to a universal physical clock. The systems and techniques described in this document align events dynamically without the need for a universal physical clock.

A node that orders the events, which can also be referred to as an upward node, can determine a disentangled time for each event based on a time at which the node receives an event fragment for the event and age of the event indicated by an event fragment that is sent according to the protocol. The sending node can determine the age of the event based on a difference between a send time for the event fragment and either a time at which the sending node received the event, if received from another node, or a time at which the event occurred at the sending node if that is where the event occurred. The sending node can generate and send an event fragment that includes data indicating the age of the event. Additional techniques can be used to accurately account for various sources of latency in a distributed computing system. The upward node can arrange the events in an order based on the determined disentangled time for each event.

According to some implementations, a method includes receiving, by an upward node from one or more nodes of a distributed computing system, event fragments for corresponding events that have occurred within the distributed computing system. Each event fragment includes an age parameter that indicates an age of the corresponding event from a perspective of a node that sent the event fragment to the upward node. The upward node calculates, for each event, a corresponding disentangled time based on a received time that represents a time at which the upward node received the event fragment for the event and the age of the event indicated by the age parameter for the event. The upward node arranges the events in an order according to the corresponding disentangled time for each event. The age for each event that occurred at the node that sends the event fragment for the event to the upward node is equal to a difference between a send time at which the node sends the event fragment for the event to the upward node and a time at which the event occurred at the node and/or the age for each event that occurred at a different node that is different from a given node that sends the event fragment for the node to the upward node can be equal to a difference between (i) a send time at which the given node sends the event fragment for the event to the upward node and (ii) a timestamp for the event. Other implementations of this aspect include corresponding apparatus, systems, and computer programs, configured to perform the aspects of the methods, encoded on computer storage devices.

These and other implementations can each optionally include one or more of the following features. Some aspects include performing one or more actions based on the events arranged in the order. The disentangled time for each event can be equal to a difference between the received time for the event fragment for the event and the age of the event. The age for each event can be calculated by the node that sends the event fragment for the event to the upward node.

The timestamp for the event can be received from the different node and is determined by the different node based on a time at which the event occurred at the different node.

In some aspects, the age parameter for each event is selected from multiple possible ages. At least one of the possible ages can be based on an age of one or more other events. The age parameter for each event is based on a minimum send time for sending event fragments.

The methods in accordance with the present disclosure can include any combination of the aspects and features described herein. That is, methods in accordance with the present disclosure are not limited to the combinations of aspects and features specifically described herein, but also may include any combination of the aspects and features provided.

Particular embodiments of the subject matter described in this specification can be implemented to realize one or more of the following advantages. Using the disentanglement techniques described in this document, a node of a distributed computing system can accurately order events that occurred within the distributed computing and multi-agent systems without the need for a universal physical clock and without having to synchronize clocks at the various nodes within the system. This reduces complexities and computational costs required to implement universal physical clocks, e.g., implementing and maintaining the clock itself, coupling the clock to each node, and ensuring continuous communication between the clock and each node. This also reduces the amount of data that has to be transmitted between nodes to maintain synchronization between the clocks of the various nodes. The techniques described in this document enables an upward node to determine causal relationships between events and perform causal ordering without a universal physical clock and without clock synchronization. Causal ordering is an important methodology in distributed systems for nodes in the system to agree on the sequence of events or operations. Causal event ordering is defined as sorting the events in the order of their creation from the perspective of consumer nodes, e.g., an upward node. Accurately ordering events using the described techniques prevents errors that can occur due to inaccurate ordering.

<FIG> shows an example distributed computing system <NUM> in which a node <NUM> orders events. The distributed system <NUM> includes the upward node <NUM> and additional nodes <NUM> that are connected via a data communication network, such as a local area network (LAN), a wide area network (WAN), the Internet, a mobile network, or a combination thereof.

The node <NUM> that orders the events can be referred to as an upward node that receives data related to events, e.g., in the form of event fragments, from the other nodes <NUM>. In some implementations, the node <NUM> is a consumer node that consumes events and performs actions based on the events, or another node in the network that is tasked with event ordering. Each node <NUM> and <NUM> can be a computer or other type of device that sends and receives data in the distributed computing system <NUM>. The nodes <NUM> process events and send event data for events, e.g., in the form of event fragments, to other nodes <NUM> in the distributed computing system <NUM>. At least some of the nodes <NUM> send event fragments to the upward node <NUM> For ease and clarity of subsequent description, event data is described as being sent in the form of event fragments and an example event fragment is shown in <FIG> and described below. However, event data can be sent in other forms as well.

The upward node <NUM> includes an event ordering engine <NUM> that is configured to determine the order of the events corresponding to the event fragments based on data in the event fragments. The upward node <NUM> can store the event fragments and data that indicates the order of the events in episodic memory <NUM>. The episodic memory <NUM> can be in the form of one or more hard drives, cloud computing data storage, flash drives, and/or other types of data storage devices.

In some implementations, the event ordering engine <NUM> can order events based on an assumption that there is no latency, e.g., no delay or uncertainty, in transmitting event fragments. That is, there is an assumption that there are no network delays in transmitting event fragments between nodes, no application delays, and no waiting at nodes, e.g., no waiting for the event data for events to processed at the nodes. In some implementations, the event ordering engine <NUM> orders events using techniques that account for latency, e.g., the various delays and uncertainty in the distributed computing system <NUM>. Determining the order of events can include determining a causal event order by sorting the events in the order of their occurrence (e.g., their creation) from the perspective of the upward node <NUM>.

<FIG> are diagrams that show some assumptions that are made in some no latency approaches described in this document. A first assumption is that there is no delay in the system and a second assumption is that there is no time dilation in the system, or the time dilation is very negligible if any. The approaches can be used in implementations in which there are latencies, but the approaches are used to order events under the assumption that there are no latencies or time dilations or that the latencies and time dilations are negligible. In general, time dilation is a difference in elapsed time measured by two different clocks, e.g., clocks of different nodes.

<FIG> is a diagram that shows an example event sequence <NUM> for events that occur at nodes of a distributed system. In this example, it is assumed that there is no delay or uncertainty in any format in the distributed computing system. For example, this assumes that there are no delays in sending data over the network, no delays caused by applications, and no delays waiting for nodes to send data that is ready to be sent. If there is an event at node m, the appropriate event fragment is created at the same time that the event occurs and the event fragment broadcasted to neighbor nodes, e.g., node l, at the same time that the event occurs. Node l will also receive the event fragment at the same time that the event occurs. In this example, em, z represents event #z at node m and rtl, em,z represents the received time for an event fragment for event #z at node l. In this example, the two times are equal since there is no latency in transmitting the event fragment for the event. In other words, the event fragment for event #z was received at node l at the same time that event #z occurred and at the same time that the event fragment for event #z was sent from node z.

Similarly, when an event occurs at node l, node l can send an event fragment to node m with event data for the event. For example, el, y+<NUM> represents event #y+<NUM> at node l and rtl, el,y+<NUM> represents the received time for an event fragment for event #y+<NUM> at node m. Again, the two times are equal since there is no latency in transmitting the event fragment for the event.

<FIG> is a diagram that shows equal time differences <NUM> for multiple nodes of a distributed system. That means the duration between time t2 and time t1 (e.g., t2-t1) in all nodes including the upward node should be equal or at least close to equal, e.g., within a threshold such that the difference is negligible. In other words, there is no time dilation, or it very negligible, if any. In this example, the calculated age time, as described below, in any node is equal to a duration of time (in the upward node perspective), from the time the event occurred to the received time at the upward node (UP). The received time at the upward node is the time at which the event fragment for the event is received at the upward node.

<FIG> shows the structure of an example event fragment <NUM>. The event fragment <NUM> can have a particular structure that corresponds to a temporal disentanglement protocol used by nodes of distributed systems to send event data to other nodes. A node can send an event fragment for an event to other nodes, e.g., to neighboring nodes that neighbor the node in the distributed system. One or more nodes can also send event fragments to an upward node that determines the order of the events corresponding to the event fragments.

The event fragment <NUM> includes an operation domain action pair data field <NUM>, an origin data field <NUM>, a sequence data field <NUM>, an age data field <NUM>, a geolocation metadata field <NUM>, additional customizable metadata fields <NUM>-<NUM> - <NUM>-N, and payload data <NUM>. The operation domain action pair field <NUM> can include data identifying the operation of the event and the domain on which the event occurred. The origin data field <NUM> can include origin data that identifies the node at which the event occurred, e.g., using a node identifier for the node. The sequence data field <NUM> includes sequence data that can be used to arrange the event corresponding to the event fragment <NUM> in a sequence with other events. In some implementations the sequence data for an event is a combination of a timestamp (e.g., in milliseconds) and Lamport clock, LP. The age data field <NUM> includes age data, e.g., in the form of an age parameter, that represents the age of the event at the time the event fragment <NUM> is sent to the upward node from the perspective of the node that sends the event fragment <NUM> to the upward node.

Each Node Ni in a distributed system maintains a monotonically increasing sequence Seqi which is a combination of timestamp Ci (in milliseconds) and Lamport clock, LP. This sequence Seqi can be included in the sequence data field <NUM> of the event fragment. The timestamp can be based on a current time of a clock of the node that is sending the event fragment. The Lamport clock is an algorithm used to determine the order of events in a distributed system. For example, in a multi-process distributed environment, each process can have a single timestamp counter. The value of this counter is incremented before assigning it to each new event. When the event is received by another process, the recipient's counter is updated to the maximum value between its current counter and the received timestamp counter. The resulting number is then incremented by a value of one. This approach resolves same-millisecond collisions between different processes.

When an event z occurs at Node Ni, the node Ni can set the sequence Seqi(e) to be the time of the current timestamp Ci + LP of zero. The event z can be represented as eNi, z to identify the node Ni and the event z at the node Ni. If there is more than one event that occurred with the same timestamp Ci, the Lamport clock LP will be increased by one for each event that occurred with the same timestamp Ci. A node sends the event's eNi, z fragment to the (e.g., selected list of) neighbor nodes, e.g., using multicast. As shown in <FIG>, each event fragment includes an origin, an operation, and a sequence - Seqi (eNi, z). When the event fragment is received at another node, Nj, the node Nj will record the received time, e.g., the local time based on a local clock of node Nj, of the fragment. This received time can be represented as rtNj, eNi,z to identify the node Nj and the event eNi, z.

The node Nj adds the received event fragments that it received from neighbor nodes, along with its own events, to a queue, QNj,k. The node Nj can periodically, or when able, send the event fragments stored in the queue QNj,k to the upward node. The parameter k is used to show kth time to add the received event fragments and events into the queue QNj,k and broadcast the event fragments to the upward node. Just before sending the event fragments in the queue to the upward node, the node Nj can calculate the age for all events that have an event fragment in the queue QNj,k. The node Nj can add the age for an event to the age date field <NUM> of its event fragment before sending the event fragment <NUM> to the upward node.

The send time for an event can be represented as stNj, k, which indicates that the send time is at kth time and is for an event fragment sent by the node Nj. A node, e.g., node Nj, can determine the age for an event based on the send time for its event fragment and either (i) the time the event fragment was received if the event fragment was received from another node or (ii) the timestamp for the event if the event occurred at the node that is determining the age of the event. For node Nj, the age for events of event fragments received from node Ni can be determined using Equation <NUM> below: <MAT>.

In Equation <NUM>, AgeeNi,z represents the age of event z that occurred at node Ni, stNj, k represents the send time at which node Nj is sending the event fragment for event z at node Ni, and rtNj, eNi,z represents the received time that node Nj received the event fragment for event z from node Ni.

For node Nj, the age for its own events, e.g., events that occurred at node Nj, can be determined using Equation <NUM> below: <MAT>.

In In Equation <NUM>, AgeeNj,y represents the age of event y that occurred at node Nj, stNj, k represents the send time at which node Nj is sending the event fragment for event y at node Nj, and Cj, eNj,y represents the timestamp for event y as determined by node Nj, e.g., using a local clock of node Nj.

To calculate the age AgeeNj,y for event y, node Nj can use the timestamp Cj, eNj,y part of its full sequence which contains the timestamp of the creation of the event, e.g., in milliseconds. The Lamport Time can be used in addition to the timestamp when there is more than one event at the same millisecond, so that the events can be differentiated.

The upward node calculates the disentangled time (DT) for each event fragment according to the upward node time, e.g., using a local clock of the upward node, and the received time of the event fragment in the send queue of the upward node. The received time for an event fragment can be represented as <MAT>, which indicates that the received time is based on the upward node's clock and is for an event fragment received from node Nj. The upward node can determine the disentangled time for the event corresponding to the event fragment using Equation <NUM> below: <MAT>.

In Equation <NUM>, DTex represents the disentangled time for an event corresponding to event fragment x, <MAT> represents the received time for this event fragment, and Ageex represents the age for the event corresponding to the event fragment. This calculation is possible due to the assumption that there is no (or little) time dilation between all nodes in the distributed system. The upward node arranges the events in order based on their disentangled times. If the disentangled time DTeNi,z for event z at node Ni is less than the disentangled time DTeNj,y for event y at Node Nj, then event z at node Ni (eNi,z) happened before event y and Node Nj (eNj,y). If the disentangled times are equal (DTeNi,z = DTeNj,y), then the events eNi,z and eNj,y happened at the same time.

<FIG> shows an example data flow <NUM> for receiving an event fragment and determining the age for events at the recipient node. The data flow <NUM> is shown using an example architecture of fragment broadcasting between nodes K and L.

Node K sends an event fragment <NUM> for an event Fx that occurred at node K. The event fragment <NUM> can include the same structure as the event fragment <NUM> of <FIG>. When node L receives the event fragment <NUM>, node L can store the event fragment <NUM> in local memory <NUM> and records a received time <NUM> for the event fragment <NUM> in local memory. The received time can be the local time based on a local clock of node L. Node L can also place the event fragment in a queue <NUM> of event fragments to be sent to an upward node that can determine the disentangled time for each event and store the event fragments in episodic memory <NUM>.

Node L maintains the queue <NUM> of event fragments and calculates the age of the events for the event fragments when propagating the event fragments to an upward node, e.g., an upward node that updates the episodic memory <NUM>. The environment <NUM> can include additional nodes that send event fragments to node L and/or to other nodes that use a queue to propagate event fragments to an upward node.

Node L can use an algorithm <NUM> to determine the age for each event for which an event fragment is stored in the queue <NUM>. If the event fragment is a sibling event fragment received from another node, Node L can determine the age for the event of the event fragment using the send time at which Node L is sending the event fragment to the upward node and the received time for the event fragment, e.g., using Equation <NUM> above. If the event fragment is for an event that occurred at Node L, Node L can determine the age for the event of the event fragment using the send time at which Node L is sending the event fragment to the upward node and the timestamp for the event, e.g., using Equation <NUM> above. After determining the ages, Node L can add the ages to their respective event fragments and send the event fragments to the upward node and/or to episodic memory <NUM>.

<FIG> is a diagram that shows an example event sequence <NUM> for events that occur at nodes of a distributed system. This figure illustrates a case in which Node l is broadcasting its send queue <NUM> with two event fragments for events of its own (el,y and el,y+<NUM>) and two event fragments received from Node m, (event fragments for events em,z and em,z+<NUM>) to an upward node. At the upward node, the upward node can determine the disentangled time (DT) of each event fragment using the received time (rtM,upl,k) for the events in the send queue <NUM> and their respective ages. Two noteworthy parts of each element of the table in <FIG> are the origin (e.g., the node at which the event occurred) and the sequence. The payload part can be empty or null until the upward node receives the event fragment and its payload from the originating node of that event.

The following discussion evaluates the impact of fragment-event broadcasting ratio and multicast factor on network load and n-guarantee. The n-guarantee is defined as a minimum number "n" of event fragments needed in the upward node to calculate the disentangled time with desired, defined, and/or target confidence. For the purpose of this evaluation, the following parameters are defined in Table <NUM>:.

<FIG> shows an example sequence <NUM> of event fragments broadcast by a node, Node i. A number of event fragments <NUM>, e.g., event fragments <NUM>-<NUM> to <NUM>-<NUM>) can be generated and broadcasted by Node i equal to ri + (fi * ri * mi), which includes ri events per second and fi ri mi event fragments broadcasted by Node i to mi neighbor nodes. Each neighbor node will receive a given event fragment from Node i with a probability of prob_fi. In this example, the number of event fragments that will be added to the send queue of neighbor nodes by Node i is equal to: prob_fi * fi * ri * mi.

The same technique can be used to evaluate the number of elements (e.g., event fragments) added to the send queue of Node i per second assuming Node i has mi neighbors. In this example, the number elements added to the send queue of Node i per second can be determined using Equation <NUM> below: <MAT>.

Having prob_upi and knowing that Node i will broadcast to the upward node every <MAT> second, the number of elements (e.g., event fragments) will be added to the table and/or episodic memory of the upward node by Node i per second can be estimated using Equation <NUM> below: <MAT>.

Finally, the total number of elements (e.g., event fragments) expected to be received at the upward node from all nodes in the network per second can be estimated using Equation <NUM> below: <MAT>.

<FIG> shows an example environment <NUM> in which an upward node <NUM> orders events of event fragments received from nodes <NUM>. To support n-guaranteed, there should be a minimum number of neighbor nodes to receive the event fragments from a specific node, e.g., Node i. For example, to have <NUM>-guranteed, at least one of the neighbor nodes should receive the event fragment (Fi,<NUM>) from Node i with target confidence (Pr(Fi,<NUM>)) determined using Equation <NUM> below: <MAT>.

Considering the example environment <NUM> of <FIG>, Pr(Fi,<NUM>) of <NUM> can be evaluated if probability of a successful communication between Node i and its neighbor nodes (A, B and C) is <NUM>, which can be determined using Equation <NUM> below: <MAT>.

In the case of <NUM>-guranteed, there should be at least two of neighbor nodes {l, m, n} of Node i that receive the fragment (Fi,<NUM>). Pr(Fi,<NUM>) of <NUM> can be evaluated using Equation <NUM> below: <MAT>.

The probability of <NUM>-guranteed and <NUM>-guranteed can be estimated at the upward node which means at least one or two of the neighbor nodes received the fragment (Fi. ,<NUM>) and has/have successful upward broadcast. These probabilities can be estimated using Equations <NUM> and <NUM> below: <MAT> <MAT>.

In this section, latency and delay in a distributed system and how such latency is accounted for when ordering are described. The latency can be from networking (e.g., transmitting data across a network), buffering event data and/or event fragments, application delays, and/or other latency sources. If there is an event at Node m, the appropriate fragment will be created and broadcast to neighbor nodes, e.g., Node l. The event fragment will be received at Node l with some delay Δ. The same assumption applies to the upward broadcasting of the send queue to the upward node as well.

In general cases, it was assumed that all elements of the send queue will be broadcasted at the same time, e.g., the queue of Node m with stm, i-<NUM>, and will be received by upward node at the same time, rtM, upm,i-<NUM>. However, there could be some use cases and/or applications in which the send time and receive time are unique to each event fragment in the send queue. These assumptions are shown in <FIG>, which shows an example event sequence <NUM> for events that occur at nodes of a distributed system.

Referring to <FIG>, Node m sends an event fragment for event z, em,z. Rather than being received by Node l at the same time that event z occurred at Node m, the event fragment for event z is received at Node l at time rtl, em,z after some time delay <NUM>. In addition, each event fragment of a send queue <NUM> of Node l is received by the upward node after some time delay <NUM> after being sent from Node l.

<FIG> is a diagram that shows an example event sequence <NUM> for events that occur at nodes of a distributed system. The event sequence <NUM> shows a case in which an event fragment for event z that occurred at Node m, em,z, was created at Node m and was broadcast to the upward node. There is going to be latency, Δupm,i-<NUM>, from the send time at Node m, stm, i-<NUM>, until it received at upward node, rtM, upm,i-<NUM>, considering all various types of delay and latency in the distributed system. Note that Δupm,i-<NUM> could be different for each element, e.g., event fragment, of the send queue. In an ideal case, the age of event em,z should be as determined using Equation <NUM> below: <MAT>.

In Equation <NUM>, stm, i-<NUM> represents the time that Node m sends the event fragment for event z and Cm, em,z represents the timestamp for event z.

In this example, the disentangled time (DT) of event z, em,z, at the upward node can be determined using Equations <NUM> or <NUM> below: <MAT> <MAT>.

In Equations <NUM> and <NUM>, rtM, upm,i-<NUM> represents the received time at which the upward node receives the event fragment for event z and Age represents the age for event z, which can be determined using Equation <NUM> as shown in Equation <NUM>.

To account for latency in the distributed system, the upward node can determine the age and disentangled time using Equations <NUM>-<NUM> below: <MAT> <MAT> <MAT>.

In Equation <NUM>, stm, i-<NUM> represents the send time at which Node m sends, to the upward node, an event fragment for event z, Cm, em,z represents the timestamp for event z as determined by Node m, and Δupm,i-<NUM> represents the latency between the send time and the received time at which the upward node receives the event fragment for event z from Node m. The other parameters are the same as those in Equations <NUM> and <NUM>. In this example, there would be a distance of Δupm,i-<NUM> from ideal DT in the computed DT that accounts for latency.

<FIG> is a diagram that shows an example event sequence <NUM> for events that occur at nodes of a distributed system. The event sequence <NUM> of <FIG> shows a case in which event z of Node m, em,z, is created and Node m sends, e.g., via multicast, the relevant event fragment for event z to Node l. Later, Node l broadcasts all elements of its send queue including the event fragment for event z, em,z, to the upward node. In this case, there will be two latencies associated with this event, em,z. One latency <NUM>, represented as Δl, em,z, is the time difference between when Node m supposedly sends the event fragment, from the perspective of Node l, and the received time at which of the event fragment is received by Node <NUM>, which is represented as rtl, em,z. Another latency <NUM>, represented as Δupl,k, is the time difference from the send time at Node l, represented as stl, k in the perspective of the upward node, until the event fragment is received at the upward node, represented as rtM, upm,i-<NUM>, considering all various types of delay and latency in the distributed system.

Similarly, if there is no latency, the ideal age of em,z is shown in Equation <NUM> below: <MAT>.

In Equation <NUM>, stl, k is the send time for the event fragment in the perspective of the upward node and rtl, em,z is the received time at which the event fragment is received at Node <NUM>.

The actual age <NUM> and disentangled time (DT) <NUM> for event z, em,z, when accounting for latency are represented by Equations <NUM>-<NUM> below: <MAT> <MAT> <MAT>.

In Equation <NUM>, the actual age is represented as the ideal age plus both sources of latency described above. In Equations <NUM> and <NUM>, rtM, upl,k is the received time at which the upward node receives the event fragment and the age is based on Equation <NUM>. In this example, inaccuracy in the disentangled time DT calculation will be the sum of the delays, Δl, em,z + Δupl,k, as the exact value of these parameters are unknown.

The following description defines example methods and techniques that reduce the impact of those latencies when determining the age and disentangled time for events in a distributed system. One or more of the methods can be used in any combination to account for latency in the distributed system.

In a first method, the impact of latency on the age of event fragments at the node level is reduced by finding the best age for each event fragment. This provides a more accurate disentangled time that reduces the impact of latencies in determining the disentangled time.

In <FIG>, which shows an example event sequence <NUM> for events that occur at nodes of a distributed system, it can be seen that Node m is sending, e.g., multicasting, three event fragments for three events em,z, em,z+<NUM> and em,z+<NUM>, out of six total events (fm=<NUM>/<NUM>) to Node l.

The time difference between each pair of events, d<NUM>, d<NUM> and d<NUM> can be defined using Equations <NUM>-<NUM> below: <MAT> <MAT> <MAT>.

In these equations, the time difference d<NUM> is the time difference between events em,z and em,z+<NUM>, the time difference d<NUM> is the time difference between events em,z+<NUM> and em,z+<NUM>, and the time difference d<NUM> is the time difference between events em,z and em,z+<NUM>. Each time difference can be determined based on the difference between the timestamps Cm for the two events, as shown in Equations <NUM>-<NUM>.

The age of event em,z can be computed in three ways. One example way is using its received time of the fragment at Node l and the send time of the kth element in the queue to the upward node using Equation <NUM> below: <MAT>.

In Equation <NUM>, ageem,z represents the age of event em,z, stl, k represents the send time of the kth element in the queue to the upward node, and rtl, em,z is the received time at which Node l received the event fragment for event em,z.

Two additional ways of determining the age of event em,z use the ages of events em,z+<NUM> and em,z+<NUM>, and time differences d<NUM> and d<NUM>, respectively, as shown in Equations <NUM> and <NUM> below: <MAT> <MAT>.

In <FIG>, a<NUM>, a<NUM> and a<NUM> represent ageem,z, ageem,z+<NUM> and ageem,z+<NUM>, respectively. That is, a<NUM>, a<NUM> and a<NUM> represent the ages of events em,z, em,z+<NUM> and em,z+<NUM>, respectively. In Equation <NUM>, the age of event em,z is determined based on the sum of the age of event em,z+<NUM>, represented as ageem,z+<NUM>, and the time difference d<NUM> between events em,z and em,z+<NUM>. The age of event em,z+<NUM> can be determined based on the send time stl, k of the kth element in the queue to the upward node and the received time rtl, em,z+<NUM> of the event fragment for the event em,z+<NUM> at which Node l received the event fragment for event em,z+<NUM>.

In Equation <NUM>, the age of event em,z is determined based on the sum of the age of event em,z+<NUM>, represented as ageem,z+<NUM> and the time difference d<NUM> between events em,z and em,z+<NUM>. The age of event em,z+<NUM> can be determined based on the send time stl, k of the kth element in the queue to the upward node and the received time rtl, em,z+<NUM> of the event fragment for the event em,z+<NUM> at which Node l received the event fragment for event em,z+<NUM>.

At the upward node, the disentangled times (DTs) of all events corresponding to the event fragments can be computed using the determined ages. As described above, the determined ages can be included in the event fragments for the events. Because of an unknown amount of latency, there is going to be some distance from ideal disentangled times. The upward node can determine the disentangled times for each event using Equations <NUM>-<NUM> below. To the right of each equation, the distance from the ideal disentangled time for the event is show. <MAT> <MAT> <MAT>.

In Equation <NUM>, the disentangled time DTem,z for event em,z is determined based on the received time rtM, upl,k that the upward node received the event fragment for em,z based on the upward node's clock and the age a<NUM> for the event em,z. The distance from the ideal disentangled time for the event em,z is the sum of the delay between the time that Node m sends the event fragment for the event em,z to Node l and the time that Node l received the event fragment, which is represented as Δl, em,z, and the delay between the time that Node l sends the event fragment for the event em,z to the upward node and the time that the upward node receives the event fragment for the event em,z, which is represented as Δupl,k.

In Equation <NUM>, the disentangled time DTem,z+<NUM> for event em,z+<NUM> is determined based on the received time rtM, upl,k that the upward node received the event fragment for em,z+<NUM> based on the upward node's clock and the age a<NUM> for the event em,z+<NUM>. The distance from the ideal disentangled time for the event em,z+<NUM> is the sum of the delay between the time that Node m sends the event fragment for the event em,z+<NUM> to Node l and the time that Node l received the event fragment, which is represented as Δl, em,z+<NUM>, and the delay between the time that Node l sends the event fragment for the event em,z+<NUM> to the upward node and the time that the upward node receives the event fragment for the event em,z+<NUM>, which is represented as Δupl,k.

In Equation <NUM>, the disentangled time DTem,z+<NUM> for event em,z+<NUM> is determined based on the received time rtM, upl,k that the upward node received the event fragment for em,z+<NUM> based on the upward node's clock and the age as for the event em,z+<NUM>. The distance from the ideal disentangled time for the event em,z+<NUM> is the sum of the delay between the time that Node m sends the event fragment for the event em,z+<NUM> to Node l and the time that Node l received the event fragment, which is represented as Δl, em,z+<NUM>, and the delay between the time that Node l sends the event fragment for the event em,z+<NUM> to the upward node and the time that the upward node receives the event fragment for the event em,z+<NUM>, which is represented as Δupl,k.

By having the disentangled time (DT) for any events, coming from the same send queue in this case, the disentangled time of the rest of the events from the same node can be calculated. As the upward node has a sequence of all events (e.g., as received in the event fragment as described above), subsequently their time stamp C and time difference between each pair of events, the upward node can calculate potential disentangled times DTs for event em,z and then compare them with the original disentangled time for event em,z and use the best (which may be the earliest) disentangled time. For example, depending on which disentangled time(s) the upward node has, the upward node can use one or more of the following three relationships to determine any missing disentangled times and determine the best disentangled time for the events.

In the first relationship, it is assumed that the upward node has determined the disentangled time DTem,z+<NUM> for event em,z+<NUM>. In this example, the upward node can determine the disentangled time DT'em,z+<NUM> for event em,z+<NUM> as the difference between the disentangled time DTem,z+<NUM> for event em,z+<NUM> and the time difference d<NUM> between events em,z+<NUM> and em,z+<NUM>, which can be determined using Equation <NUM> above. Similarly, the upward node can determine the disentangled time DT'em,z for event em,z as the difference between the disentangled time DTem,z+<NUM> for event em,z+<NUM> and the time difference d<NUM> between events em,z and em,z+<NUM>, which can be determined using Equation <NUM> above.

In the second relationship, it is assumed that the upward node has determined the disentangled time DTem,z+<NUM> for event em,z+<NUM>. In this example, the upward node can determine the disentangled time DT"em,z+<NUM> for event em,z+<NUM> as the sum of the disentangled time DTem,z+<NUM> for event em,z+<NUM> and the time difference d<NUM> between events em,z+<NUM> and em,z+<NUM>, which can be determined using Equation <NUM> above. Similarly, the upward node can determine the disentangled time DT"em,z for event em,z as the difference between the disentangled time DTem,z+<NUM> for event em,z+<NUM> and the time difference d<NUM> between events em,z and em,z+<NUM>, which can be determined using Equation <NUM> above.

In the third relationship, it is assumed that the upward node has determined the disentangled time DTem,z for event em,z. In this example, the upward node can determine the disentangled time DT‴em,z+<NUM> for event em,z+<NUM> as the sum of the disentangled time DTem,z for event em,z and the time difference d<NUM> between events em,z and em,z+<NUM>, which can be determined using Equation <NUM> above. Similarly, the upward node can determine the disentangled time DT‴em,z+<NUM> for event em,z+<NUM> as the sum of the disentangled time DTem,z for event em,z and the time difference d<NUM> between events em,z and em,z+<NUM>, which can be determined using Equation <NUM> above.

If it is assumed that Δl, em,z+<NUM> < Δl, em,z < Δl, em,z+<NUM>, this would result in: <MAT>.

The upward node can select the best disentangled time for event em,z based on the three disentangled times shown above. In some implementations, the upward node selects, as the best disentangled time, the earliest disentangled time among the three possible disentangled times. The upward node can then use the selected disentangled time to determine the disentangled times for the other events using the relationship corresponding to the selected disentangled time for event em,z. For example, if the upward node selects disentangled time DTem,z, the upward node can use the third relationship to determine the disentangled times for events em,z+<NUM> and em,z+<NUM>.

In a second example method, the impact of upward broadcasting latency on disentangled times (DTs) is reduced by finding a minimum average time of upward broadcasting at the node level. With the time passing and/or increment of number of upward broadcasting, it will get closer to the ideal disentangled time DT for each event.

In <FIG>, which is a diagram <NUM> that shows example time delays for transmitting event fragments between nodes, two upward broadcasting of Node l's queue (Q) <NUM>, at a kth time (<NUM>-<NUM>) and at a k+<NUM>th time (<NUM>-<NUM>) is shown. Note that this method considers b ms as buffer time before sending the queue and Node l will add any new event fragments to the next sending queue.

First, the upward node calculates the average time of upward broadcasting each individual element of the send queue <NUM>, which is represented as avg_send_time. The parameter Size_Q is the number of elements, e.g., event fragments, in the queue <NUM>. Then, the upward node compares the average send time with the best (e.g., smallest) average time of previous iterations and selects the minimum one as the new minimum send time which is represented as min_send_time.

When evaluating the age of each element before upward broadcasting to the upward node, a fraction of min_send_time is added to the age. If the event fragment j if received from another node, e.g., Node Ni, Equation <NUM> can be used to determine the age of the event. If the event occurred at Node <NUM>, then Equation <NUM> can be used to determine the age of the event. <MAT> <MAT>.

In Equation <NUM>, the age of the event is determined based on the send time stl, k of the event fragment for the event, the received time <MAT> of the event fragment for the event, an impact factor, α, the size of the queue Size_Q, and the minimum send time (min_send_time). In Equation <NUM>, the age of the event is determined based on the send time stl, k of the event fragment for the event, the timestamp Cl, el, y for the event, the impact factor, α, the size of the queue Size_Q, and the minimum send time (min_send_time).

The impact factor, α, represents how much of the minimum send time (min_send_time) will be added to the age of event fragments and events and it could be defined in various methods based on the use cases and application. Here, one example implementation is being shown. It would improve and get closer to ideal age by time passing and/or the number of upward broadcasting, Upcount. For example, in this case kth, k+<NUM>th and. β is a constant value which determines how fast α convergence to <NUM> as shown in Equation <NUM> below: <MAT>.

<FIG> shows example pseudocode <NUM> for determining average and minimum send times at the node level. In this example pseudocode, the size (Size_Q) of the send queue is initialized to the length (Len_Q) of the send queue. The age of an event in the queue is evaluated using the send time (send_time), the buffer time (b), the minimum send time (min_send_time), the impact factor (α), and the size (Size _Q) of the send queue. For example, the pseudocode could use Equations <NUM> and <NUM> above to determine the age of each event in the send queue.

A time parameter is initialized to the current time and the elements, e.g., event fragments, in the send queue are sent to the upward node. The average send time (avg_send_time) is determined based on the current time, the time parameter, and the size (Size_Q) of the send queue. The minimum send time is determined based on the minimum value of the minimum send time (min_send_time) and the average send time (avg_send_time).

In a third method, the disentangled time (DT) of events is improved by finding the best age at the upward node. This is achieved by analyzing various ages for each event coming from an origin node and its neighbor nodes.

<FIG> is a diagram that shows an example event sequence <NUM> for events that occur at nodes of a distributed system. In this figure, an event em,z created at Node m and an event fragment for the event em,z have been received at Node l and Node n with Δl, em,z and Δn, em,z latency, respectively.

All three nodes Node m, Node l, and Node n send, e.g., broadcast, their queues including the event fragment for event em,z, which will be received at the upward node with Δupm,i-<NUM>, Δupl,k and Δupn,a latencies, respectively. A combination of these latencies will create three potential disentangled times for em,z: DTem,z, DT'em,z and DT"em,z, which can be defined using the following relationships: <MAT> <MAT> <MAT>.

The upward node can analyze the disentangled times of em,z in order of their reception at the upward node to choose the best (e.g., earliest) one. If there was any for which the disentangled time needs to be updated, not only is em,z updated, but also previous events' disentangled times that came from the same node, here Node m, and before em,z, e.g., em,z-<NUM> , em,z-<NUM> and so on, are also updated.

<FIG> shows a flow chart of an example process <NUM> for arranging events in order. The process <NUM> can be performed by a node, e.g., an upward node, of a distributed system that includes multiple nodes communicable coupled to each other, e.g., via a network. Operations of the process <NUM> can also be implemented as instructions stored on one or more computer readable media, which may be non-transitory, and execution of the instructions by one or more data processing apparatus can cause the one or more data processing apparatus to perform the operations of the process <NUM>. For brevity, the example process <NUM> is described in terms of being performed by an upward node.

The upward node <NUM> receives event fragments for events that have occurred within the distributed system (<NUM>). Each event fragment includes, for example, an age parameter that indicates an age of the corresponding event from a perspective of a node that sent the event fragment to the upward node. Each event fragment can include additional data, as described above, e.g., with reference to <FIG>.

The age for each event can be calculated by the node that sends the event fragment for the event to the upward node. This node can calculate the age using different techniques. For example, the age can be calculated in different ways depending on whether the event occurred at the node calculating the age or another node that sent an event fragment to the node performing the calculations. In a particular example, the age for each event that occurred at the node that sends the event fragment for the event to the upward node can be equal to a difference between (i) a send time at which the node sends the event fragment for the event to the upward node and (ii) a time at which the event occurred at the node, e.g., using Equation <NUM>. The age for each event that occurred at a different node that is different from a given node that sends the event fragment for the node to the upward node can be equal to a difference between (i) a send time at which the given node sends the event fragment for the event to the upward node and (ii) a timestamp for the event, e.g., using Equation <NUM>. The timestamp for the event is received from the different node and is determined by the different node based on a time at which the event occurred at the different node.

In some implementations, the age parameter can be determined in ways that account for latency within the distributed system. For example, the age parameter for each event can be selected from multiple possible ages where least one of the possible ages is based on an age of one or more other events, e.g., using Equations <NUM>-<NUM>. The age parameter for each event can be based on a minimum send time for sending event fragments, e.g., using Equations <NUM> and <NUM>.

The upward node calculates a disentangled time for each event (<NUM>). The disentangled time for each event can be based on a received time that represents a time at which the upward node received the event fragment for the event and the age of the event indicated by the age parameter for the event, as described above. The disentangled time for each event can be equal to a difference between the received time for the event fragment for the event and the age of the event. Any of the techniques described above can be used to calculate the disentangled time for each event.

The upward node arranges the events in an order according to the corresponding disentangled time for each event (<NUM>). For example, the upward node can put the events in order from the lowest disentangled time to the highest disentangled time.

The upward node can also perform actions based on the events and/or their order. For example, some events and/or particular sequences of events can trigger actions by the upward node.

Embodiments of the subject matter described in this specification can be implemented as one or more computer programs, i.e., one or more modules of computer program instructions encoded on a tangible non-transitory program carrier for execution by, or to control the operation of, data processing apparatus.

Computers suitable for the execution of a computer program include, by way of example, general or special purpose microprocessors or both, or any other kind of central processing unit.

In some embodiments, a server transmits data, e.g., an HTML page, to a user device, e.g., for purposes of displaying data to and receiving user input from a user interacting with the user device, which acts as a client. Data generated at the user device, e.g., a result of the user interaction, can be received from the user device at the server.

An example of one such type of computer is shown in <FIG>, which shows a schematic diagram of a computer system <NUM>. The system <NUM> can be used for the operations described in association with any of the computer-implemented methods described previously, according to one implementation. The system <NUM> includes a processor <NUM>, a memory <NUM>, a storage device <NUM>, and an input/output device <NUM>. Each of the components <NUM>, <NUM>, <NUM>, and <NUM> are interconnected using a system bus <NUM>. The processor <NUM> is capable of processing instructions for execution within the system <NUM>. In one implementation, the processor <NUM> is a single-threaded processor. In another implementation, the processor <NUM> is a multi-threaded processor. The processor <NUM> is capable of processing instructions stored in the memory <NUM> or on the storage device <NUM> to display graphical information for a user interface on the input/output device <NUM>.

Claim 1:
A method performed by one or more data processing apparatus (<NUM>; <NUM>; <NUM>), the method comprising:
receiving, by an upward node (<NUM>; <NUM>) from one or more nodes (<NUM>; <NUM>) of a distributed computing system (<NUM>), event fragments (<NUM>; <NUM>) for corresponding events that have occurred within the distributed computing system, wherein each event fragment comprises an age parameter that indicates an age of the corresponding event from a perspective of a node that sent the event fragment to the upward node;
calculating, by the upward node and for each event, a corresponding disentangled time based on (i) a received time (<NUM>) that represents a time at which the upward node received the event fragment for the event and (ii) the age of the event indicated by the age parameter for the event; and
arranging, by the upward node, the events in an order according to the corresponding disentangled time for each event
wherein the age for each event that occurred at the node that sends the event fragment for the event to the upward node is equal to a difference between (i) a send time at which the node sends the event fragment for the event to the upward node and (ii) a time at which the event occurred at the node; and/or wherein the age for each event that occurred at a different node that is different from a given node that sends the event fragment for the node to the upward node is equal to a difference between (i) a send time at which the given node sends the event fragment for the event to the upward node and (ii) a timestamp for the event.