Elastic reply-request multicast messaging protocol for peer-to-peer distributed systems

Techniques disclosed herein provide a messaging protocol for a distributed system. In one embodiment, each message constructed according to the messaging protocol includes a field for a list of globally unique identifiers (GUIDs) of nodes that should not reply to the message. A new node joining the system sends a message requesting system state information and including an empty GUID list, indicating that all nodes receiving the message should reply. In normal operation, the system's state information is synchronized among all nodes at regular intervals by exchanging messages indicating local state changes at each node, and including GUID list fields specifying all other nodes' GUIDs so that no nodes reply. A node which identifies messages that should have been received from another node but are missing transmits a message requesting the missing messages, with a GUID list including all nodes other than the other node that sent the missing messages.

BACKGROUND

A distributed system is a system in which components execute concurrently to achieve a common goal. The components of distributed systems typically communicate through message passing. Maximizing the performance of such communication can be important for efficiently achieving the common goal of the distributed system.

The most common network configurations in distributed systems are master/slave and peer-to-peer configurations. In a master/slave configuration, one node acts a master node by establishing timing and controlling communications with slave nodes, typically through a request-and-reply model (or a multicast request followed by individual replies). However, in such a configuration, the slave nodes are unable to initiate communications with the master node or with each other, and the master node's failure affects overall system communication, i.e., the master node is a single point of failure.

By contrast, nodes in a peer-to-peer configuration are each able to initiate communications with other nodes when there is a need for data exchange, and such communications may be through the request-and-reply model or with all peers through a multicast model. Although the peer-to-peer configuration does not have a single point of failure, use of the request-and-reply model may result in heavy network traffic that decreases communication performance. For example, to synchronize data among N nodes, each of the N nodes needs to transmit a request for information to the other nodes, and each of the N nodes also needs to reply to requests from N−1 other nodes, giving a total of N requests+N*(N−1) replies, or N2, messages. Multicast may be used to address this performance issue with the request-and-reply model by sending messages to a group of recipients at a time. However, traditional multicast does not guarantee the delivery of messages or their delivery order, so data may not be successfully synchronized among the nodes of the distributed system.

SUMMARY

One embodiment provides a method for a node in a distributed system to process a message. The method generally includes receiving the message, where the message includes a request and identifiers of all nodes in the distributed system from which replies are not required. The method further includes, responsive to determining that the received message does not include an identifier associated with the node: processing the request, and transmitting a reply to the request.

Further embodiments include a non-transitory computer-readable storage medium storing instructions that when executed by a computer system cause the system to perform the method set forth above, and a system programmed to carry out the method set forth above.

DETAILED DESCRIPTION

Embodiments presented herein provide an elastic reply-request multicast (ERRM) messaging protocol for a peer-to-peer distributed system. Messages constructed according to such a protocol are elastic in the sense that the messages are forwarded to all nodes in the system but replies are required only as needed. This elastic messaging approach has the multicast model's advantage of sending messages to groups of nodes, but also the request-and-reply model's advantage of guaranteeing delivery and ordering of messages, by requiring those nodes which have not previously replied to a request or which have sent out-of-order messages to reply to the current message.

In one embodiment, each message includes a field reserved for a list of globally unique identifiers (GUIDs) of nodes that do not need to reply to the message. Upon joining the distributed system, a new node sends a message with (1) a request for system state information needed to initialize the new node, and (2) an empty GUID list, indicating that all nodes receiving the message should reply. A message broker forwards such a message to all other nodes in the system, which respond with locally maintained state information that is then used by the new node to obtain the current system state. Further, the new node identifies the GUIDs of responding nodes (from, e.g., a GUID field in the reply message) and uses the identified GUIDs to construct the GUID list field of subsequent messages so that nodes that provide requested information are not required to reply again to retry messages. After new nodes are initialized, the distributed system may enter normal operation in which the system state is synchronized among all nodes at regular intervals by exchanging messages that indicate local state changes at each of the nodes. Such exchanged messages may include GUID list fields specifying all other nodes' GUIDs, making the messages essentially multicast messages that do not require replies from any receiving nodes. The normal operation may be interrupted when a node detects that it is missing messages from other nodes (e.g., based on an out-of-order numbering in received messages that should be consecutively numbered), and the detecting node may then send, for each such other node, a message requesting the missing message(s) from that other node, with a GUID list including all nodes other than that other node.

FIG. 1illustrates components of a system100in which an embodiment may be implemented. As shown, system100is a peer-to-peer distributed system with nodes110in communication with each other via a message broker120. Although four nodes1101-4are shown, it should be understood that distributed system100may include any number of nodes110. Clients140communicate with nodes110via a network (e.g., the Internet) by invoking APIs130(via, e.g., TCP messages) to request services from nodes110. Illustratively, client requests are routed through message broker120, which may also serve as a load balancer that distributes client requests to appropriate nodes110.

In one embodiment, nodes110may contribute substantially equally to achieving a common computing goal, such as servicing clients140. Each of nodes110further maintains local state data that is periodically exchanged with every other node110in distributed system100to synchronize the overall system state data. For example, the state data may include counts of the number of times clients have been serviced, and such counts may be exchanged in every predefined time interval so that each node110knows the total count of times that each client has been serviced by distributed system100. The total count for a particular client may then be compared to a quota set for the client to determine whether the client is permitted to request further services from distributed system100in a given time period. Although counts are used herein as a reference example, it should be understood that any other type of data may also be exchanged among nodes110. It is assumed herein that only eventual state data consistency is required, i.e., that distributed system100is able to tolerate transient data inconsistencies during predefined time intervals.

Each of nodes110represents a server computer (or another type of computer) constructed on a conventional hardware platform. The hardware of such a server (or other) computer may include CPU(s), a memory, network interface controller(s) (NICs), and an I/O device interfaces, among other things. Although discussed herein with respect to nodes110communicating with each other using an ERRM messaging protocol, it should be understood that such communications may actually be performed by applications running in the server (or other) computers, or any other feasible software or hardware component of the server (or other) computers. Similarly, message broker120may be a server (or other) computer running an application (or other software or hardware) that routes messages between nodes110, as well as between nodes110and clients140. In another embodiment, message broker120may run in one of the nodes110. In yet other embodiments, nodes110may include virtualized computing systems hosting virtual machines (VMs) or containers that service requests from clients140, or the nodes110themselves may be VMs or containers.

Nodes110are configured to subscribe to topics with message broker120, indicating their interest in receiving messages having those topics. In turn, message broker120forwards messages having each topic to nodes which have subscribed to the topic. For example, a topic may be subscribed to by all nodes110in distributed system100. Such a topic may be used by new nodes to send messages to all other nodes110requesting system state information needed for initialization, and such a topic may also be used by nodes110in normal operation to send messages at regular intervals to distribute local state information to other nodes110for synchronization purposes, and by nodes110to request missing messages that have been sent by other nodes but not yet received.

In one embodiment, messages may be constructed according to the ERRM protocol. As previously noted, such messages are elastic in that replies are required only as needed. Each message may include a field reserved for a list of GUIDs of nodes that should not reply to the message. As discussed in greater detail below, a new node joining distributed system100may send a message requesting system state information and including an empty GUID list, indicating that all nodes receiving the message should reply with their locally maintained state information. In normal operation after new nodes have joined distributed system100, the system's state information may be synchronized among all nodes110by exchanging messages at regular intervals, with the messages indicating local state changes at each node and including GUID list fields specifying all other nodes' GUIDs so that no other nodes reply to the messages. In addition, a node which identifies messages that should have been received but are missing may transmit, for each node which sent missing message(s), a respective request for the missing message(s) sent by that node and including a GUID list specifying all nodes other than that node.

FIG. 2illustrates contents of an ERRM message200, according to an embodiment. As shown, message200includes a subject field210, a serial number field220, a GUID list field230, a request field240, a local changes field250, and a reply subject field260. An example of such a message200in JavaScript Object Notation (JSON) format is as follows:

{“subject”:“emitter.data.inbound.>”,“message_serial_number”:“CC4EE3AF921F0E1A_00000111”,“GUIDList”:{“DF3TY2QJ068D1T4U”,“YR2HH6PO57W33TT”,.....,“RBN345UTOK374TM”},“request”:“emit data map of the whole system”,“local changes”:“localEmitCount=2515, .....”,........ ,“reply subject”:“CC4EE3AF921F0E1A.reply”}
It should be understood that message formats other than JSON may also be used.

Subject field210specifies the subject of the message, which may be one of a number of topics that nodes110in distributed system100have subscribed to. As discussed, message broker120is configured to forward messages to all nodes that have subscribed to topics indicated in the messages. In the JSON example above, the subject is “emitter.data.inbound.>,” and all nodes subscribed to “emitter.data.inbound.>” will receive the message.

Serial number field220includes the sending node's GUID and a message number, which may be a unique number incremented with every message sent by the node. In the JSON example above, the sending node's GUID is CC4EE3AF921F0E1A, and the message number is 00000111. A receiving node may use the message numbers of messages it receives to identify out-of-order messages. Each of nodes110is configured to maintain the highest message number previously received from each other node. Each node may then identify messages that are received out of order by comparing the highest previous message number from a particular node with a current message number. For example, if the highest previous message number received from node CC4EE3AF921F0E1A is 00000109, then receipt of the message from node CC4EE3AF921F0E1A with message number 00000111 would be out of order, as message number 00000110 has yet not been received. In such a case, the receiving node may transmit an ERRM message that lists all GUIDs except CC4EE3AF921F0E1A and a request for message number 00000110. Although discussed herein primarily with respect to maintaining the highest previous message number from each other node, in alternative embodiments, other message history may also be maintained.

GUID list field230includes a list of GUIDs belonging to nodes which should not reply to message200. Nodes which receive message200and find their own GUID in the GUID list230do not process the request in message200. For example, nodes having GUIDs DF3TY2QJ068D1T4U, YR2HH6PO57W33TT, and RBN345UTOK374TM in the JSON example above would not process the message. Conversely, nodes which do not find their own GUIDs in the GUID list230do process message's200request. It should be understood that elasticity of message200is controlled via GUID list field230. When GUID list field230is empty, all recipients of message200need to respond with a reply message. At the other extreme, when GUID list field230specifies GUIDs of all nodes110in distributed system100, then message200is essentially a multicast message that does not require replies from any receiving nodes.

Request field240specifies the type of information requested from the nodes which receive message200and are not included in GUID list field230. The requested data may include, e.g., state information maintained by all other nodes in the case of a new node joining the distributed system110(e.g., “emit data map of the whole system” in the JSON example above), or information in missing messages that need to be resent in the case of out-of-order messages.

Local changes field250specifies changes to the system state maintained by the sender of message200. As discussed, nodes110are configured to exchange messages, such as message200, that indicate local state changes at each of the nodes110. When distributed system100is in normal operation, such messages may be transmitted by each node during every time interval (e.g., every minute) so as to synchronize the state information at the nodes110. Aside from such messages, other messages, such as those requesting initialization data and missing messages, may also include local changes. In the JSON example above, local changes field250includes a “localEmitCount,” which is a count of the number of times a particular client has been serviced. As discussed, distributed system100may employ such counts to track how much of each client's quota has been filled. For example, each client may be an application with a respective quota specifying an allowed number of service request fulfillments for a given time period, and the count for that application may be used to determine whether more requests may be serviced under the quota.

Reply subject field260is similar to subject field210and specifies the subject that a reply to message200should use. Nodes which receive message200and do not find their own GUID in GUID list230may transmit reply messages with subject fields210specifying the reply subject field260topic. In the JSON example above, a reply message may specify the CC4EE3AF921F0E1A.reply topic, which the sending node CC4EE3AF921F0E1A is subscribed to. As a result, message broker120may then forward such a reply message to the node CC4EE3AF921F0E1A.

FIG. 3illustrates a method300for sending request messages, according to an embodiment. As shown, method300begins at step310, where node110icreates a GUID list. As discussed, GUID lists may be created to store GUIDs of nodes that previously provide requested information and are not required to reply again to the same request message. In the case of a new node joining distributed system100, an empty GUID list may be created, as the new node is not yet aware of other nodes110in distributed system100. This permits new nodes, including nodes that have gone offline (e.g., for maintenance or as a result of a crash) and are rejoining distributed system100, to request and receive system state information without having advanced knowledge of other nodes110in distributed system100. Allowing new nodes to be network agnostic in this manner improves system extensibility, as new nodes may be added to distributed system100relatively easily.

Conversely, a GUID list may be created that includes all other nodes' GUIDs when distributed system100is in normal operation, during which nodes110periodically exchange local state changes without expecting replies. Continuing the example from above in which system data needs to be synchronized among N nodes, only N messages, each having all GUIDs of other nodes in its GUID list, need to be sent in each time interval. This is in contrast to the N2messages required in the case of request-and-reply. It should be understood that network efficiency is improved, as the distributed system may often run in the normal operating mode.

In addition, when messages are received out of order from one or more other nodes110, indicating that intervening messages are missing, then a respective message may be constructed for each of those nodes with a GUID list that includes GUIDs of all nodes other than one of the nodes from which messages are received out of order. In such a case, only the node that sent out-of-order messages and is not in the GUID list is required to reply with the missing message or messages. This approach overcomes limitations of the multicast communication model, in which there are no guarantees of message delivery or delivery order. Essentially, ERRM messages are multicast messages, but with the added ability to request specific information from specific peer nodes. Based on the message numbers discussed above, nodes110may identify received messages that are out of order and request missing messages to be resent, thereby improving reliability of distributed system100. Although discussed herein with respect to requesting missing messages, it should be understood that no such requests may be made in cases where, e.g., the missing messages are not actually needed.

At step320, node110isends a message with the GUID list. If node110iis a new node joining distributed system100, then node110imay transmit a message that includes a request for system state information needed to initialize node110iand an empty GUID list indicating that all nodes receiving the message should reply. In such a case, the request for initialization information may be included in request field240, and GUID list field230may include the empty GUID list. On the other hand, if node110iis an existing node and distributed system100is in normal operation, then node110imay periodically send a message with its local state changes and a GUID list that includes GUIDs of all other nodes110. When messages are received from another node out of order, node110imay send a message with a request for missing message(s) in request field240and GUIDs in GUID list field230of all nodes other than the other node from which messages are received out of order. In one embodiment, node110imay send a separate message with one missing GUID (and all other GUIDs listed) to each node from which missing message(s) are requested, so that the particular node with the missing GUID in each such message replies to the request. It may also happen that more than one node sent missing messages with the same serial numbers, in which case node110imay send a single message requesting messages with those serial numbers and specifying all GUIDs of nodes other than those which sent the missing messages.

In one embodiment, the message sent at step320may also include local state change information added to local changes field250. Such local state changes may be sent with every message so that nodes110in distributed system100are able to maintain synchronized state information of the overall distributed system100. When requesting missing messages or initialization information, the local state change information is included to essentially piggyback off of the request so as to also transmit local state change information. In one embodiment, the distributed system100may employ counts to track how much of clients' service request quotas have been filled, and messages may be sent with state change information indicating the locally maintained counts after clients are serviced at the local node110i. Such local counts after clients are serviced by node110iwill differ from the counts maintained by other nodes110, and each of the nodes110may use the counts received from other nodes to update their locally maintained count information of other nodes.

At step330, node110ireceives reply message(s). In the case of a new node joining distributed system100, message broker120is configured to forward the message sent at step320to all other nodes110in distributed system100, which may then respond with reply messages including state information that they maintain. It should be understood that some of the state information received from different nodes may be the same, but there is no harm to receiving duplicate data. In the case of messages received out of order and requests being made for missing messages, other nodes110that receive such requests and are not listed in GUID list field240may reply with the missing messages. No replies are required when distributed system100is in normal operation, in which each node broadcasts messages with local state changes and all other nodes' GUIDs in GUID list field240.

At step340, node110iadds GUIDs from the reply message(s) to the GUID list created at step310. In the case of a new node joining distributed system100and expecting reply messages including state information from every other node110, when receiving replies, the new node builds its GUID list by adding GUIDs extracted from serial number field220of all reply messages to the GUID list. In the case of messages received out of order from another node and a request being made for the missing message(s), the GUID of the other node that is required to reply with the missing message(s) is first removed from the GUID list. Then, when the missing message(s) are received as a reply or replies, node110iadds the GUID extracted from serial number field220of the received reply or replies back to the GUID list.

In one embodiment, node110imay maintain in its memory a data structure that keeps track of which nodes have replied to which types of requests, and such a data structure may be used to determine the GUIDs that need to be added to or removed from the GUID list. In another embodiment, node110imay also extract the message number from serial number field220of reply messages and keep track of the highest message number from each other node110so that out-of-order messages may be identified.

At step350, node110idetermines whether to resend the message sent at step320. For a new node joining distributed system100, node110imay resend the message requesting system state information (e.g., a predefined number of times), in case any of the other nodes110did not receive the previously sent message. For messages received out of order and a message being sent requesting missing messages, node110imay need to resend the message with requests for messages that are still missing, from nodes that do not reply to the initial message requesting the missing messages.

If node110idetermines at step350that the message does not need to be resent, then method300ends. Otherwise, if node110idetermines that the message needs to be resent, then method300returns to step320, where node110isends a message with the GUID list. It should be understood that the GUID list has been updated at step340to include GUIDs of nodes that previously replied. As a result, those nodes that previously replied will find their GUIDs in the GUID list and will not reply again when they receive the later message.

FIG. 4illustrates a method400for processing a received message, according to an embodiment. As shown, method400begins at step410, where node110idetermines whether the received message was sent by itself. As discussed, message broker120forwards messages to all nodes in distributed system100that have subscribed to the messages' topics. Node110imay subscribe to some of the same topic(s) as the messages it sends. In such cases, node110imay determine, based on the GUID in serial number field220, that it is the sender of a received message and ignore the message at step420.

If node110iinstead determines that it was not the sender of the received message, then at step430, node110iprocesses the information in local changes field250of the received message. As discussed, each message may include local state change information so that nodes110in distributed system100are able to maintain synchronized state information (e.g., synchronized counts tracking how much of clients' service request quotas have been filled, etc.) of the overall distributed system100. Local state information may change as clients are serviced, and each node110adds such local state information that has changed to messages sent to other nodes110so that those other nodes110may update their own state information.

At step440, node110idetermines if the message number in serial number field220is out of order. As discussed, each of the nodes110keeps track of the highest message number previously received from each other node so that later out-of-order messages may be identified. Node110imay determine that the message number in serial number field220is out of order if a comparison of this message number with the highest previous message number from the same sending node indicates that one or more messages have message numbers between the highest previous message number and the message number in serial number field220. As messages that are sent should be consecutively numbered, the messages having message numbers between the highest previous message number and the message number in serial number field220are messages that have been sent but not received by node110i, i.e., the messages are missing.

If node110idetermines at step440that the message number in serial number field220is out of order, then at step450, node110iremoves the GUID of the message's sender from a GUID list created by node110i, discussed above with respect to method300. In addition, node110iprepares a request for the missing message(s) to be included in subsequent message(s) sent by node110i, the subsequent message(s) including the GUID list with the GUID of the message's sender removed. In one embodiment, node110imay send a separate message with one missing GUID (and all other GUIDs listed) to each node from which missing message(s) are requested, so that the particular node with the missing GUID in each such message replies to the request. If more than one node sent missing messages with the same serial numbers, then node110imay send a single message requesting messages with those serial numbers and specifying all GUIDs of nodes other than those which sent the missing messages.

At step460, node110idetermines if its own GUID is included in the GUID list field240of the received message. As discussed, GUIDs in GUID list field240indicate nodes110that should not reply to the received message. If node110idetermines its own GUID is included in GUID list field240of the received message, then node110itakes no further action.

On the other hand, if node110idetermines that its own GUID is not included in GUID list field240of the received message, then at step470, node110iprocesses the request in request field240of the received message. As discussed, the request may be, e.g., a request for a missing message or a request for all local state information needed for initialization. At step480, node110isends a reply which includes the requested information.

Advantageously, techniques disclosed herein provide an ERRM messaging protocol in which messages are forwarded to all nodes in the distributed system, but replies are required only as needed by excluding GUIDs of nodes which need to reply from the messages' GUID lists. The disclosed approach has the multicast model's advantage of sending messages to groups of nodes, thereby reducing network traffic. This is particularly the case during normal operation in which each node periodically sends a message to other nodes with a GUID list specifying all of the other nodes, none of which are required to respond. When system data needs to be synchronized among N nodes, only N such messages need to be sent per time interval, rather than the (approximately) N2messages as in the case of request-and-reply. The ERRM messaging protocol also has the request-and-reply model's advantage of guaranteeing delivery and ordering of messages by requiring specific nodes (e.g., nodes which have not previously replied to a request or nodes that sent missing messages) to reply to the current message. This is achieved by removing those nodes that are required to respond from the GUID list specifying nodes that do not need to respond. Further, as the ERRM messaging protocol uses peer-to-peer network configurations in which each node is equal to other nodes, there is no single point of failure as in master/slave network configurations. Nodes that go down (e.g., for maintenance purposes or as a result of a crash) may already have transmitted their local state information to other nodes in the distributed system, after which those nodes may reacquire the system state information when rejoining the distributed system as new nodes. In addition, nodes seeking to join the system may be network agnostic and simply send a request message with an empty GUID list, making it relatively easy to add new nodes.

In addition, while described virtualization methods have generally assumed that virtual machines present interfaces consistent with a particular hardware system, persons of ordinary skill in the art will recognize that the methods described may be used in conjunction with virtualizations that do not correspond directly to any particular hardware system. Virtualization systems in accordance with the various embodiments, implemented as hosted embodiments, non-hosted embodiments, or as embodiments that tend to blur distinctions between the two, are all envisioned. Furthermore, various virtualization operations may be wholly or partially implemented in hardware. For example, a hardware implementation may employ a look-up table for modification of storage access requests to secure non-disk data.