JP7765348B2 - Fault-tolerant system and data processing method - Google Patents

Description

The present invention relates to a multi-processing system that applies fault-tolerant system technology.

A fault-tolerant system is a computer system that performs the same processing across multiple computers (nodes) and allows operations to continue uninterrupted even if a failure occurs on one of the computers.

State machine replication is known as one approach to achieving fault-tolerant systems. In state machine replication, the same input is provided to each replicated node, which has the same state, and processing is performed deterministically using that input and state. As a result, each node produces the same output and is updated to the same state. This means that even if a failure occurs in some nodes, the remaining nodes can continue operating without inconsistencies.

Here, in the case of an application configured to execute multiple tasks in parallel, the overall behavior can be made consistent between nodes by making the order in which shared state is accessed consistent.

For example, in the data processing method disclosed in Patent Document 1, the leader node communicates the access order to the follower nodes each time it accesses shared data. Meanwhile, the follower nodes wait for instructions from the leader node regarding the access order each time they access the shared data, and process the data in that order. In this way, the data processing method of Patent Document 1 ensures that the behavior of replicated nodes is consistent.

International Publication No. 2012/127652

However, in the above-mentioned Patent Document 1, determinism is guaranteed by matching the order of access to shared data between tasks between nodes each time an access is made. In other words, with the data processing method of Patent Document 1, communication occurs between the leader node and follower node each time an access request is made to data shared between tasks, which could increase the load for synchronization depending on the workload.

The present invention was made in consideration of the above points, and aims to provide a low-latency, scalable fault-tolerant system and data processing method that can harmonize the behavior between nodes while reducing the load caused by synchronization.

In order to solve this problem, the present invention provides a fault-tolerant system in which multiple nodes are connected via a communication network, each of which has a processor, memory, and storage area. The processor in each node reads a program from the storage area into the memory and executes it. The application is comprised of one or more tasks that deterministically generate an output for an input. One of the multiple nodes is a leader node, and the other nodes are follower nodes. The leader node and the follower nodes are each configured to execute the same task. Each task has a status indicating the state of the task. A fault-tolerant system is provided in which a task is linked one-to-one to a local state, the task inputs include messages received from outside the system or from other tasks, and access to the local state linked to the task, and the task outputs include messages sent by the task to outside the system or to other tasks, and updates to the local state linked to the task, and when the task sends a message to a destination task having multiple message senders, the processors of the multiple nodes reach a consensus among the multiple nodes on the order in which messages should be delivered to the destination task, and send the messages according to the agreed-upon delivery order.

In order to solve this problem, the present invention provides a data processing method using a fault-tolerant system in which multiplexed nodes are connected via a communication network, each of the multiple nodes having a processor, memory, and storage area, and an application that runs in each of the nodes by the processor reading a program from the storage area into the memory and executing it, and the application is composed of one or more tasks that deterministically generate output for an input, one of the multiple nodes is a leader node, and the nodes other than the leader node are follower nodes, and the leader node and the follower node are each configured to execute the same task, and each of the tasks is A data processing method is provided in which a task is linked one-to-one to a local state that indicates the state of the task itself, the task's input includes messages received from outside the system or from other tasks, and access to the local state linked to the task, the task's output includes messages sent by the task to outside the system or to other tasks, and updates to the local state linked to the task, and when the task sends a message to a destination task having multiple message senders, the processors of the multiple nodes reach a consensus among the multiple nodes on the order of message delivery to the destination task, and send the messages according to the agreed-upon delivery order.

The present invention provides a low-latency, scalable fault-tolerant system and data processing method that can harmonize the behavior of nodes while reducing the load caused by synchronization.

1 is a block diagram showing an example of the configuration of a fault-tolerant system (FT system) 1 according to an embodiment of the present invention. FIG. 1 is an image diagram showing the overall characteristics of processing when an event occurs in the FT system 1. FIG. 10 is an image diagram illustrating the correlation between nodes. FIG. 10 is a diagram illustrating the characteristics of a message. FIG. 10 is an image diagram showing an example of the operation of task 20. FIG. 10 is a diagram showing an example of a message path management table 35. FIG. 10 is a diagram for explaining characteristics of consensus building on an event log between nodes. 10 is a flowchart illustrating an example of a processing procedure of a leader in consensus building of an event log. 10 is a flowchart illustrating an example of a processing procedure of a follower in consensus building of an event log. 10 is a flowchart showing an example of a procedure for a receiving process by an input processing unit 31 of a reader. 10 is a diagram showing an example of the data structure of an external input type log entry stored in the event log 34. FIG. 10 is a flowchart illustrating an example of a processing procedure for a message request acceptance process. FIG. 10 illustrates an example of a message delivery type log entry. FIG. 10 is a model diagram for explaining the relationship between nodes when a message path is added. 10 is a flowchart illustrating an example of a processing procedure for a message path addition process. FIG. 10 illustrates an example of a message path addition type log entry. FIG. 10 is a diagram illustrating another example of a log entry of the message path addition type. FIG. 10 is a diagram illustrating an example of message path information. 10 is a flowchart illustrating an example of a processing procedure for message path information reception processing. FIG. 10 is a diagram showing a specific example of a message path management table 35. FIG. 10 illustrates an example of a log entry of a message path information type. FIG. 10 is a sequence diagram illustrating an example of a processing procedure for adding a message path according to a second improvement method. 10 is a flowchart illustrating an example of a processing procedure for message path addition processing according to a second improvement method. 10 is a flowchart illustrating an example of a processing procedure for message pass information reception processing according to a second improvement method. FIG. 10 is a conceptual diagram illustrating order assurance of external outputs. FIG. 10 is a conceptual diagram illustrating a configuration in which external outputs from a plurality of nodes are adjusted and output.

Embodiments of the present invention will be described in detail below with reference to the drawings.

The following description and drawings are illustrative of the present invention, and have been omitted or simplified as appropriate for clarity of explanation. Furthermore, not all of the combinations of features described in the embodiments are necessarily essential to the solution of the invention. The present invention is not limited to the embodiments, and all application examples that conform to the concept of the present invention are included within the technical scope of the present invention. Those skilled in the art will be able to make various additions and modifications to the present invention within the scope of the present invention. The present invention can also be implemented in a variety of other forms. Unless otherwise specified, each component may be plural or singular.

In the following explanation, various types of information may be described using terms such as "table," "list," and "queue," but the various types of information may also be expressed using data structures other than these. To indicate that the information is not dependent on the data structure, "XX table," "XX list," etc. may be referred to as "XX information." When describing the content of each piece of information, terms such as "identification information," "identifier," "name," "ID," and "number" are used, but these terms are interchangeable.

In addition, in the following explanation, when describing elements of the same type without distinguishing between them, reference signs or common numbers in reference signs will be used, and when describing elements of the same type with distinction between them, the reference signs of those elements will be used, or the ID assigned to those elements will be used instead of the reference signs.

In addition, while the following description may describe processing performed by executing a program, a program is executed by at least one processor (e.g., a CPU) to perform a predetermined process using storage resources (e.g., memory) and/or interface devices (e.g., communication ports) as appropriate, and therefore the processor may be the entity performing the processing. Similarly, the entity performing the processing by executing a program may be a controller, device, system, computer, node, storage system, storage device, server, management computer, client, or host having a processor. The entity performing the processing by executing a program (e.g., a processor) may include a hardware circuit that performs some or all of the processing. For example, the entity performing the processing by executing a program may include a hardware circuit that performs encryption and decryption, or compression and decompression. A processor operates as a functional unit that realizes a specified function by operating in accordance with a program. Devices and systems that include a processor are devices and systems that include these functional units.

A program may be installed on a device such as a computer from a program source. The program source may be, for example, a program distribution server or a computer-readable storage medium. When the program source is a program distribution server, the program distribution server includes a processor (e.g., a CPU) and storage resources, and the storage resources may further store a distribution program and a program to be distributed. The processor of the program distribution server may then execute the distribution program, causing the processor of the program distribution server to distribute the program to be distributed to other computers. Also, in the following description, two or more programs may be realized as one program, and one program may be realized as two or more programs.

(1) Configuration Fig. 1 is a block diagram showing an example of the configuration of a fault-tolerant system (FT system) 1 according to one embodiment of the present invention. As shown in Fig. 1, the FT system 1 is a computer system made up of multiple computers (nodes 10) connected via a communication network 2.

The FT system 1 according to this embodiment is a so-called leader-follower distributed system, in which one of the multiple nodes 10 serves as a leader node (also called the leader), and the nodes other than the leader node serve as follower nodes (also called followers). Specifically, consider the leader and follower in the RAFT distributed consensus algorithm. In the FT system 1, the same input is given to an event-driven application replicated on multiple nodes 10, and the same output is obtained. An event-driven application is composed of tasks 20, which are multiple processing modules capable of operating independently. In other words, the FT system 1 is configured so that when an event occurs, the same task 20 is executed on both the leader node and the follower node.

While the FT system 1 is operating normally, the leader node 10 and follower node 10 will not switch positions. However, if, for example, there is a possibility that a failure has occurred in the leader node 10, that node 10 will be disconnected from the FT system 1, and one of the follower nodes 10 will become the new leader, and the FT system 1 will continue processing. This configuration allows for the realization of a scalable FT system 1.

The FT system 1 is also connected to one or more terminals 3 via a communications network 2, and transmits and receives data to and from the terminals 3. The communications network 2 is, for example, a local area network (LAN), but is not limited to this. The terminals 3 are computers used by users, and, for example, act as clients to send events to the FT system 1 and receive the results of event processing from the FT system 1. As shown in Figure 2, which will be described later, in the FT system 1, the leader node is the first to receive an event.

Node 10 is a computer equipped with a processor such as a CPU (Central Processing Unit), memory, a storage device (storage area), a communication interface, etc. The storage device stores various programs and data, such as the event processing platform 30 and tasks 20, that are executed by node 10. The processor then reads these various programs from the storage device, expands them into memory, and executes them.

A task 20 is an independently operable event-driven processing module (program) that constitutes an application with a group of multiple tasks on one or more nodes 10. Specific examples of a task 20 include a microservice, a function on a FaaS (Function as a Service), or a process on an OS (Operating System).

In this embodiment, the application state is managed independently for each task 20 by a state (local state) 21. The state 21 is accessed exclusively. Each task 20 is associated one-to-one with the state 21 that indicates the state of the task itself, and operates in loosely coupled cooperation with other tasks 20. In other words, each task 20 does not have directly shared application data, and when accessing application data (state 21) managed by another task 20, this is done exclusively by a unidirectional message (hereinafter simply referred to as a message) requesting processing. Furthermore, each task 20 is configured to behave deterministically. In other words, once the input and current state are determined, the combination of the output and post-processing state is uniquely determined.

The event processing infrastructure 30 is an operation system that processes events. The event processing infrastructure 30 is capable of multitasking and can execute multiple tasks 20 in parallel. The event processing infrastructure 30 has an input processing unit 31, a message processing unit 32, an output processing unit 33, an event log 34, a message path management table 35, and a message waiting queue 36.

The input processing unit 31 receives events from the outside (terminal 3) or the leader node, and calls the appropriate task to process the events in the order in which they occur.

The message processing unit 32 accepts message requests from tasks 20 and sends messages to the target tasks 20. At this time, the message processing unit 32 manages the message path management table 35, which is necessary to ensure determinism throughout the system, and performs inter-node synchronization processing.

The output processing unit 33 arbitrates the processing results of each node 10 as necessary and ultimately transmits them to the outside. The arbitration of processing results can be achieved, for example, by transmitting the result selected by majority vote, or by transmitting on a first-come, first-served basis.

The event log 34 is a data structure that manages information for which consensus is to be reached (or for which consensus has been reached) between nodes 10. Specifically, consider the log in the distributed consensus algorithm RAFT. The event log 34 is composed of a collection of log entries. A log entry is a unit of information for which consensus is to be reached, and is first added to the log by the leader, who then replicates it to the followers. When the leader receives reports from the majority of followers that replication of the log entries is complete, the log entries are committed, starting with the oldest. The leader then notifies the followers of how many log entries have been committed. Each node 10 applies the committed log entries, starting with the oldest. How they are applied depends on the type of log entry.

The message path management table 35 is data that manages message paths. A message path is a route that connects the message sender and the message destination. Details will be described later, but the message path management table 35 is registered based on message path information received from each node, including the local node.

A message waiting queue 36 is provided for each message path. When a message request occurs, the message request is queued in the message waiting queue 36 in order. If the message waiting queue 36 is not in a blocked state, the message request is dequeued from the top of the message waiting queue 36 at a specified timing and sent to the message destination.

(2) Overview of Event Processing Figure 2 is an image diagram showing the overall processing characteristics when an event occurs in the FT system 1. The external input 38 shown in Figure 2 performs external input of an event or transfers the event to follower nodes 10B and 10C, and forms a consensus between nodes regarding the externally input event. The external input 38 specifically corresponds to the input processing unit 31 shown in Figure 1. The external output 39 outputs the processing result in the node 10 to the leader node 10A or to the outside. The external output 39 specifically corresponds to the output processing unit 33 shown in Figure 1.

As shown in Figure 2, in the FT system 1 according to this embodiment, the external input 38 of the leader node 10A first receives an event, then forwards the received event to the external inputs 38 of the follower nodes 10B and 10C, and consensus is reached between the nodes. Then, each of the nodes 10A, 10B, and 10C calls the appropriate task 20 in the order determined by the leader node 10A to process the received event.

In Figure 2, each task 20 processes received messages sequentially and deterministically. For example, if task A receives messages Ma1 and Ma2 in that order, it first processes message Ma1 and sends the resulting message Mb1 to task B, then processes message Ma2 and sends the resulting message Mb2 to task B. In this case, since task B has only a single message sender, messages Mb1 and Mb2 are processed sequentially and deterministically at every node. In this way, determinism is maintained for single-path message delivery in the FT system 1 without synchronization.

Also, as shown by the dashed line from task D to task B in Figure 2, when a message path is added from one task to another task for the first time, a consensus is reached between the nodes about the existence of that message. As a result, if multiple message senders appear on the path, a consensus is reached on the order in which messages will be processed thereafter.

Furthermore, in cases where messages are delivered via multiple paths, such as when task C operates upon receiving a message from task B or D in Figure 2, the order in which messages are processed is agreed upon between nodes. This ensures that the behavior of a task with multiple paths (for example, task C) is consistent between nodes.

After task 20 has been processed at each node 10A, 10B, and 10C, external output 39 performs arbitration such as majority voting as necessary and outputs the processing results to the outside (terminal 3).

(3) Ensuring determinism of task operation Figure 3 is an image diagram illustrating the correlation between nodes. In Figure 3, a leader node (leader) 10A is shown in the left half of the drawing, and one follower node (follower) 10B is shown in the right half of the drawing, showing the correlation between them. Note that, as will be described in detail later, various methods can be used for external output, and therefore the output processing unit 33 is illustrated without distinguishing between the leader 10A and the follower 10B. As shown in Figure 3, three types of consensus are formed between nodes: external input, message path addition, and message delivery order (task invocation order).

Figure 4 is a diagram explaining the characteristics of message delivery. As shown in Figure 4, information exchange and coordination between tasks and between external input/output and tasks is performed using messages 101. Note that messages handled in this embodiment are unidirectional; for example, a response to a message from task A to task B is treated as a message from task B to task A.

Messages flowing on a single message path never change order on that path. Each message 101 is assigned a sequence number indicating the order of transmission for each message path. In this explanation, the sequence numbers (message sequence numbers) of messages within the same message path will be represented as #1, #2, ... in the order of transmission. Message sequence numbers are managed in the message path management table 35.

For example, as shown in Figure 4, on a message path going from task B to task C, if task B sends messages #1 and #2 in that order, task C will receive the messages in the order #1 and #2. On the other hand, for messages on a message path that have the same sender but different destinations, such as message "B → External #2" going from task B to external output and message "B → C #1" going from task B to task C, the order in which they are sent does not affect which message arrives at the destination first.

As mentioned above, task 20 is an event-driven task that operates upon receiving a message 101. Each task 20 behaves deterministically and outputs a unique processing result for each input. Here, input to task 20 is the contents of message 101 received from external input 38 or other tasks 20, or the contents of its own local state 21. Furthermore, output from task 20 is the message 101 sent to other tasks 20 or external output 39, or updates to its own local state 21. Each task 20 processes one message 101 at a time. When multiple messages 101 are received, the messages 101 are processed sequentially one by one. The local state 21 is independent for each task 20 and cannot be directly accessed by other tasks 20. When a task 20 wants to access the local state 21 of another task 20, it does so by issuing a message 101 to the other task 20 requesting access.

Figure 5 is an image diagram showing an example of the operation of task 20. Figure 5(A) shows an example configuration when there are multiple message senders, and Figure 5(B) shows an example configuration when there is a single message sender. With reference to Figure 5, the principle by which the determinism of applications (specifically task operations) by each node 10 is guaranteed in this embodiment will be explained.

In Figure 5 (A), task C performs processing based on the contents of messages 101A and 101B from tasks A and B and the contents of local state 21C, updates local state 21C, and outputs the processing results to external output 39. At this time, if consensus is not reached between nodes 10 regarding the message delivery order (processing order), differences may occur in the output contents of task C and the final local state 21C between when message 101A from task A is processed first and when message 101B from task B is processed first. Specifically, for example, if message 101A from task A is processed first, task C updates local state 21C of task C with message 101A, and the contents of this updated local state 21C are used as input when processing message 101B from task B.

To resolve the above problems and guarantee determinism of processing results across the entire system (in other words, to obtain the same processing results across nodes 10), in this embodiment, in addition to tasks behaving deterministically, for tasks 20 with multiple message senders, consensus is reached between nodes 10 regarding the order in which messages should be dispatched. Details of consensus will be described later.

On the other hand, when there is only a single message source (task A), as in task C in Figure 5(B), the processing order in which message 101A is dispatched in task C is uniquely determined, as mentioned above. Specifically, in task C, message 101A is dispatched in the processing order of #3 and #4. Therefore, in the case of a single path with only a single message source, consensus building as in the case of Figure 5(A) is not necessary.

Figure 6 shows an example of the message path management table 35. The message path management table 110 shown in Figure 6 is an example of the message path management table 35 shown in Figure 1, and manages, for each message path, the sequence numbers of messages that have been dispatched on that message path. The message path management table 110 is composed of the following items: message path 111, status 112, and dispatched message sequence number 113.

Message path 111 indicates the source and destination of the message path managed by the record.

Status 112 indicates whether each message path is in a "blocked state." Specifically, for example, if the message path of the record in question is in a "blocked state," status 112 is set to "blocked." Details will be provided later in the explanation of processing by the message processing unit, but when a second message path is added when a first message path exists for a message destination task, the first message path is placed in a "blocked state," and dispatching of messages on the first message path is put on hold until consensus is reached on the second message path. The blocked state of the first message path is released when consensus is reached on the "Message Path Added" log entry regarding the addition of the second message path and this log entry is applied.

The dispatched message sequence number 113 indicates, by message sequence number, up to which message has been dispatched on the message path of the record. The message sequence number is counted for each message path, and is incremented by one each time a message occurs on that message path, and that value is assigned to that message. As a result, the message path and message sequence number can be used to uniquely identify each message.

If a pair of message destination and message source does not exist in the message path management table 110, it is treated as a new message path, and a new record is added to the message path management table 110. A new message path is added to the message path management table 110 after consensus has been reached regarding the addition of the new message path. In other words, if a message to the new message path is generated again during consensus, it is also treated as a new message path.

In addition, when registering in the message path management table 110 during actual operation of the FT system 1, it is expected that each message path will be newly registered each time a message is generated. However, as a variant, a certain number of message paths may be registered in advance. In other words, the contents of the message path management table constructed during system testing may be output to a file, and during actual operation, the contents of the output file may be loaded into the message path management table 110, so that actual operation can begin with a certain number of message paths constructed. In this way, the load on the message path registration process during actual operation can be reduced.

(4) Consensus Building Process The following describes the consensus building process between the nodes 10 in the FT system 1 according to this embodiment.

In the FT system 1, when various consensuses are reached between the nodes 10, the event log 34 (or simply the log) manages the process and results of the consensus. RAFT is a known consensus protocol, and the FT system 1 according to this embodiment essentially uses RAFT to reach consensus between the nodes 10. However, for ease of explanation, RAFT is expressed in a simplified manner in the description of this embodiment, and details such as the RAFT leader re-election protocol are omitted. Applying a log entry in RAFT means reflecting the contents of the log entry in the state machine. In this embodiment, the processing content of the log entry application differs depending on the type of log entry.

The types of log entries handled in this embodiment can be broadly divided into three categories: "external input," "message delivery," and "message path addition."

The purpose of the "External Input" log entry is to ensure that external events received by the leader are replicated to followers and dispatched in the same order at each node. The specific operation when applying the "External Input" log entry is for the message processing unit 32 to dispatch the event contents to the appropriate task for processing, depending on the contents of the external input event. An example of the data structure of an external input log entry is shown in Figure 11.

The purpose of the "Message Delivery" log entry is to ensure consistency between nodes regarding the order in which messages are dispatched when there are multiple message senders. By applying this log entry, each message is uniquely identified by the combination of the message path and message sequence number. The specific operation when applying the "Message Delivery" log entry is for the message processing unit 32 to dispatch the message. A specific example of a message delivery log entry is shown in Figure 13.

The purpose of the "Message Path Added" log entry is to ensure that nodes agree on the existence of message paths. In particular, when a second message path is added when a first message path already exists, the nodes also agree on the message sequence number up to which messages should be dispatched on the first message path before the second message path is added. Specifically, when the "Message Path Added" log entry is applied, the message processing unit 32 adds the message path to the message path management table 35 (message path management table 110 in Figure 5). Furthermore, when a first message path is added, the message processing unit 32 dispatches messages in the message waiting queue 36 of the message path in order. Furthermore, when a second message path is added, the message processing unit 32 dispatches messages in the message waiting queue 36 of the first message path in order up to the agreed-upon message sequence number, then releases the "blocked state" of the first message path and adds the second message path to the message path management table 35 (message path management table 110 in Figure 5). Specific examples of log entries for message addition paths are shown in Figures 17 and 18.

Figure 7 is a diagram illustrating the characteristics of event log consensus building between nodes. Figure 7 shows an example of the progression of log entries at different nodes 10 (leader, follower 1, follower 2).

For all three types of log entries described above, consensus on the event log 34 between nodes in this embodiment is achieved as follows in the case of Figure 7.

In Figure 7, follower 1 has completed adding log entries up to #5, and follower 2 has completed adding log entries up to #6. At this time, if the leader receives notifications from followers 1 and 2 that their entries have been added, the leader will commit the log entries up to #6, since the majority of nodes, including the leader itself, have completed adding entries up to #6, and will notify followers 1 and 2 of how many entries have been committed. Then, each node will apply any committed but unapplied log entries in order.

The detailed processing procedure for reaching consensus on such an event log 34 will be explained with reference to Figures 8 and 9.

Figure 8 is a flowchart showing an example of a leader's processing procedure for consensus building on an event log. Figure 9 is a flowchart showing an example of a follower's processing procedure for consensus building on an event log. Note that the processing shown in Figures 8 and 9 is performed by the input processing unit 31 if the log entry is of the "external input" type, and by the message processing unit 32 if the log entry is of the "message delivery" or "message path addition" type; however, for simplicity's sake, the following explanation will be described as being performed by the message processing unit 32.

As shown in FIG. 8, first, the leader's message processing unit 32 adds a log entry to the log (event log 34) (step S11). Next, the leader's message processing unit 32 requests the follower to add a log entry (step S12).

As shown in FIG. 9, when the follower's message processing unit 32 receives the request of step S12 in FIG. 8 (step S21), it adds the log entry to the log (event log 34) in the order of transmission on the message path (step S22). It then notifies the leader that the log entry was successfully added (step S23).

Next, in FIG. 8, the message processing unit 32 of the leader determines whether it has received notifications of successful addition in step S23 from a majority of the followers constituting the FT system 1 (step S13). In step S13, it determines whether the addition was successful for a majority of nodes, including the leader itself. If notifications of successful addition have been received from a majority of followers (YES in step S13), the process proceeds to step S14. On the other hand, if notifications of successful addition have not been received from a majority of followers (NO in step S13), the leader ends the processing in FIG. 8 and waits until notifications of successful addition are received from a majority of followers.

In step S14, the leader's message processing unit 32 commits the log entries for which a majority of notifications were received in step S13, in order of oldest addition. Next, the leader's message processing unit 32 notifies the follower up to which log entries (up to which log entry serial numbers) have been committed (step S15). Finally, the leader's message processing unit 32 applies the committed but unapplied log entries in order of oldest addition (step S16), and the consensus building process ends.

Meanwhile, in the follower, after notifying the leader of the successful addition of the log entry in step S23, the message processing unit 32 determines whether or not a commit notification was received from the leader in step S15 (step S24). If a commit notification was received (YES in step S24), the follower applies the committed but unapplied log entries in order from oldest to newest (step S25), and terminates the consensus building process. On the other hand, if a commit notification was not received (NO in step S24), the follower terminates the process without performing any special processing.

The above processing in Figures 8 and 9 is a flowchart in which processing is performed sequentially for ease of explanation, but this embodiment is not limited to this. For example, the consensus building process may be performed asynchronously, such as by inserting processing to wait for various messages from other nodes. Furthermore, in the following explanation, when "adding a log entry to the event log" is indicated, it is assumed that the consensus building process shown in Figures 8 and 9 is operating in the background (individual explanations will be omitted).

(5) Input processing unit 31
The following describes in detail the processing performed by the input processing unit 31. More specifically, the reception processing performed by the input processing unit 31 in response to an externally input event will be described.

Figure 10 is a flowchart showing an example of the processing procedure for reception processing by the reader's input processing unit 31. The processing shown in Figure 10 is executed, for example, periodically by the reader's input processing unit 31.

As shown in FIG. 10, the input processing unit 31 of the reader determines whether an external input event has been received (step S31). If an external input event has not been received (NO in step S31), the input processing unit 31 repeats the process of step S31 in the next cycle. If an external input event has been received (YES in step S31), the input processing unit 31 adds a log entry for the received external input event to the log (event log 34) (step S32).

Figure 11 is a diagram showing an example of the data structure of an external input type log entry stored in the event log 34. As shown in Figure 11, an external input type log entry 120 is composed of a type 121 and an external input 122. Type 121 indicates the type of log entry, and external input 122 indicates the contents of the external input. In other words, an external input type log entry stores an event received from the outside as is in the event log 34.

In step S32, the input processing unit 31 adds a log entry for the external input event to the event log 34, which initiates the consensus building process as described with reference to Figures 8 and 9, resulting in the content of the external input event being transferred from the leader to the follower. Therefore, the follower does not have a process for directly receiving the external input event as shown in Figure 10.

By performing the above-described receiving process, the input processing unit 31 ensures that the same events are dispatched in the same order on every node 10.

Then, depending on the content of the external input event, the input processing unit 31 sends a message to the appropriate task 20 that will process the event, thereby causing the event to be processed. Specifically, the input processing unit 31 sends a message request to the message processing unit 22, requesting that task 20 operate. Upon receiving this message request, the message processing unit 22 performs the message request reception process shown in FIG. 12, which will be described later, and sends a message to task 20.

As will be described later, for messages between tasks, consensus is reached between nodes to ensure that the dispatch order of those messages is consistent (consensus on message delivery order). However, messages regarding external input events are not subject to consensus on message delivery order because the dispatch order between nodes is consistent at the time the messages are sent.

(6) Message Processing Unit 32
The various processes performed by the message processing unit 32 will be described in detail below.

12 is a flowchart showing an example of the processing procedure of the message request acceptance processing. The message request acceptance processing shown in FIG. 12 is executed when the message processing unit 22 accepts a message request from the task 20.

As shown in FIG. 12, first, the message processing unit 22 determines whether the received message request requests message delivery on a new message path (step S41). Specifically, the message processing unit 22 checks whether the combination of message destination and message source indicated in the message request is registered in the message path management table 110, and can determine that it is a new message path if the message path management table 110 does not contain a pair of message destination and message source.

If the message path is a new one in step S41 (YES in step S41), the message processing unit 22 performs a message path addition process to add a new message path (step S42). Details will be described later with reference to FIG. 15, etc., but in the message path addition process in step S42, a log entry of type "Message path added" is added to the log (event log 34). At this time, as described above, the consensus building process shown in FIGS. 8 and 9 is executed in conjunction with the addition of the log entry to the log. Then, after processing step S42, the message processing unit 22 adds the accepted message request to the message waiting queue 36 (step S43) and terminates the process.

If the message path is not new in step S41 (NO in step S41), the message processing unit 22 determines whether the message from the accepted message request is a message for a task 20 with multiple message senders (step S44).

If, in step S44, the message is addressed to a task 20 with multiple message senders (YES in step S44), the message processing unit 22 adds a log entry of type "message delivery" to the log (event log 34) (step S45). After that, since the message cannot be executed until consensus is reached between the nodes, the process proceeds to the aforementioned step S43, where the accepted message request is added to the message waiting queue 36 and processing ends.

If, in step S44, the message is for a task 20 that does not have multiple message senders (i.e., is connected by a single path) (NO in step S44), the message processing unit 22 checks whether the message path is blocked (step S46). If the message path is blocked (YES in step S46), the message cannot be executed until consensus is reached between the nodes, so the process proceeds to the aforementioned step S43, adds the accepted message request to the message waiting queue 36, and terminates processing. On the other hand, if the message path is not blocked (NO in step S46), consensus is not required, so the message processing unit 22 executes the message in accordance with the accepted message request (step S47) and terminates processing.

As described above, according to the message request acceptance process shown in Figure 12, in the case of a message via a new message path or a message to a task with multiple message senders, the message will not be executed until consensus is reached between the nodes, so the message request is temporarily put on hold in the message waiting queue 36. Furthermore, for requests to an existing single message path, the message can be executed without consensus being reached as long as the message path is not blocked. Furthermore, while a second message path is being added to the destination task (while consensus is being reached between nodes), messages via the first message path are blocked.

Figure 13 shows a specific example of a message delivery type log entry. As shown in Figure 13, a message delivery type log entry 130 is composed of a type 131 and a message 132. Type 131 indicates the type of information contained in the log entry, including the message path and message sequence number. Message 132 indicates the content of each type of information indicated in type 131.

For messages addressed to a message destination with multiple message senders, a "message delivery" type log entry such as log entry 130 shown in Figure 13 is added to the log (event log 34), and the nodes reach a consensus on the order in which to dispatch the messages. Specifically, when log entry 130 shown in Figure 13 is applied, the message with message sequence number "3" on the message path "A → C" from task A to task C is dispatched to task C.

(6-2) Message Path Addition Processing The processing when a new message path is added will be described in detail below.

As mentioned above, in the FT system 1, for messages to a message destination with multiple message senders, the message dispatch order must be consistent between nodes 10. However, it may not be easy to determine in advance whether multiple message senders exist. Therefore, in this embodiment, a new message path can be added when a task 20 based on an event occurrence is executed. To prevent inconsistencies from occurring at this time, a consensus is reached so that each node 10 adds a new message path in the same state (with the same message sequence numbers to be dispatched).

Figure 14 is a model diagram that explains the relationships between nodes when adding a message path. With reference to Figure 14, we will explain in more detail the need for consensus building when adding a new message path.

Figure 14 illustrates a case where a new message path "D → B" is added when an existing message path "A → B" exists. Note that the colored messages 101 in Figure 14 represent messages that have already been dispatched. Specifically, in the case of Figure 14, node 1 (leader) has already dispatched up to #5, and node 2 (follower 1) has already dispatched up to #6.

In Figure 14, until the new message path "D → B" is added, there is only a single message path "A → B" with task B as the message destination, so messages on "A → B" are not synchronized between nodes. Therefore, when an attempt is made to add the message path "D → B," there may be a discrepancy between nodes as to how many messages have already been dispatched.

To explain this in more detail using the case in Figure 14, suppose each node adds a new message path "D→B" and dispatches messages at its own timing without reaching a consensus between the nodes. After node 1 dispatches message #5 on "A→B," a new message path "D→B" is added, message #1 on "D→B," and then messages #6 and #7 on "A→B" are dispatched in that order. In this case, the event log 34 of node 1 (the leader) records log entries in the following order: "A→B #5," "D→B Added," "D→B #1," "A→B #6," and "A→B #7." The addition of the new message path and the messages immediately preceding it (the first and second messages above) may be recorded as a single, inseparable log entry.

On the other hand, at node 2, message #6 on "A→B" has already been dispatched before the new message path "D→B" is added. Therefore, after message #6 on "A→B" is dispatched, message #1 on "D→B" is dispatched, and then message #7 on "A→B" is dispatched. At this time, the event log 34 of node 2 (follower) records log entries in the following order: "A→B #6," "D→B Added," "D→B #1," and "A→B #7." As with node 1, the addition of the new message path and the messages immediately preceding it (the first and second messages above) may be recorded as a single, inseparable log entry.

Comparing the processing results of node 1 and node 2 above, we can see that different log entries are recorded in the event log 34, and that synchronization cannot be maintained between node 1 and node 2.

In this embodiment, after adding the "D → B" message path, it is necessary to reach agreement among the nodes on the order in which the messages on "A → B" and the messages on "D → B" should be dispatched. In other words, it is necessary to reach agreement among the nodes that "after messages on the A → B message path have been dispatched up to message number X, the D → B message path will be added."

Specifically, as a protocol unique to this embodiment and not specified in RAFT, when a new message path is added, each node broadcasts message path information 102 related to the message path addition to other nodes, as shown on the right side of Figure 14. For example, the message path information 102 (node 1, A → B, 5) shown in Figure 14 indicates that a new message path was added at node 1 when message #5 was dispatched on the A → B message path. Next, based on the message path information 102 broadcast from each node, the leader determines the highest message sequence number (e.g., #6) of messages dispatched on the existing A → B message path at each node. Based on the above process, each node then agrees to add the D → B message path after the highest message sequence number (#6) dispatched on A → B, thereby achieving synchronization among all nodes.

Below, we will explain in detail the process of adding a message path (message path addition process), which triggers the above-mentioned consensus formation.

Figure 15 is a flowchart showing an example of the processing procedure for message path addition processing. The message path addition processing shown in Figure 15 corresponds to step S42 in Figure 12, and is executed by the message processing unit 22 of each node 10 when a message transmission request is received and it becomes necessary to add a new message path.

As shown in FIG. 15, first, the message processing unit 22 determines whether the newly added message path (hereinafter referred to as the new message path) is the second path for the message destination (step S51).

If the new message path is the second message path in step S51 (YES in step S51), the message processing unit 22 sets the first message path to a blocked state (step S52). Next, the message processing unit 22 broadcasts the message path information of the existing message path (see Figures 14 and 18) to other nodes (step S53), and ends the message path addition process.

In addition, each node that receives the broadcast in step S53 performs the message path information reception process shown in Figure 19, which will be described later, and the received message path information is registered in the message path management table 35, and the leader node determines the dispatch order to ensure consistency between nodes.

If the new message path is not the second message path in step S51, i.e., if it is the first or third or subsequent message path (NO in step S51), the message processing unit 22 checks whether the own node is the leader (step S54). If the own node is the leader (YES in step S54), the message processing unit 22 adds a log entry of type "Message path added" to the log (event log 34) (step S55) and ends the message path addition process. If the own node is a follower (NO in step S54), the message processing unit 22 ends the message path addition process without performing any special processing.

Figure 16 shows an example of a message path addition type log entry. Log entry 140 shown in Figure 16 is an example of a log entry that is added to the log (event log 34) in step S55 of Figure 15 when the new message path is the first or third or subsequent message path.

Log entry 140 is data managed in units of adding a new message path, and includes an existing message path 141 indicating an existing message path, an added message path 142 indicating a new message path, and a dispatched sequence number 143 indicating the sequence number of a message already dispatched on the existing message path when the new message path was added.

In the case of log entry 140 in Figure 16, the new message path is not the second message path for the message destination, so the process of broadcasting message path information for the existing message path (step S53 in Figure 15) is not performed. Furthermore, the message path information addition process in Figure 19, which is executed at each node in response to step S53 in Figure 15, is not executed, so there is no need to reference the dispatch sequence number (step S64 in Figure 19 is not performed). Therefore, in log entry 140 in Figure 16, it is not necessary to specify values for existing message path 141 and dispatched sequence number 143, and "A → B" is specified for added message path 142, indicating that the new message path is a message path from task A to task B.

Figure 17 shows another example of a message path addition type log entry. Log entry 150 shown in Figure 17 is an example of a log entry that is added to the log (event log 34) in step S64 of Figure 19 (described below) when the new message path is the first or third or subsequent message path.

Like log entry 140 in Figure 16, log entry 150 is data managed in units of the addition of a new message path, and includes an existing message path 151 indicating an existing message path, an added message path 152 indicating a new message path, and a dispatched sequence number 153 indicating the sequence number of a message already dispatched on the existing message path when the new message path was added.

Specifically, when log entry 150 in Figure 17 is applied, after messages up to message sequence number "5" are dispatched on the existing "A → B" message path, the "D → B" message path is added to the message path management table 35.

Figure 18 shows an example of message path information. Message path information is data shared between nodes to indicate how far a message has been dispatched on the first message path when a second message path to a certain message destination is added. The message path information is also used as a hint for the leader to determine when to add a message path.

The message path information 160 shown in FIG. 18 includes a node ID 161, an existing message path 162, an additional message path 163, and a dispatched sequence number 164. The node ID 161 indicates the identifier of the node 10 that is the sender of the message path information 160. The existing message path 162 indicates the sender and destination of the first existing message path. The additional message path 163 indicates the sender and destination of the message path to be added (i.e., the second message path that has the same message destination as the existing message path). The dispatched sequence number 164 indicates the maximum value of the dispatched message sequence number for the first message path indicated in the existing message path 162.

Figure 19 is a flowchart showing an example of the processing procedure for message path information reception processing. The message path information reception processing shown in Figure 19 is processing executed by the message processing unit 22 in each node 10 that receives message path information when message path information for an existing message path is broadcast in step S53 of Figure 15. In other words, the message path information reception processing is executed when the new message path is the second path for the message destination.

According to FIG. 19, first, the message processing unit 22 registers the received message path information in the message path management table 35 (step S61). A specific example of the message path management table 35 is shown in FIG. 20, which will be described later.

Next, the message processing unit 22 checks whether the node itself is the leader (step S62). If the node itself is the leader (YES in step S62), the process proceeds to step S63. If the node itself is not the leader, i.e., is a follower (NO in step S62), the message path information reception process ends.

In step S63, the message processing unit 22 of the leader determines whether message path information regarding the addition of the new message path has been received from all other nodes. If the message path information has been received from all other nodes (YES in step S63), the process proceeds to step S64. If the message path information has not been received from all other nodes (NO in step S63), the message path information reception process ends.

In step S64, the message processing unit 22 of the leader determines the maximum value of the dispatched message sequence number on the existing message paths based on the message path information from all nodes, including the leader node. It then adds a "Message path added" type log entry to the log (event log 34) with the determined maximum message sequence number as the dispatched sequence number, and terminates the message path information reception process.

Figure 20 shows a specific example of the message path management table 35. The message path management table 170 shown in Figure 20 has a message path 171 and a dispatched message sequence number 172. The message path 171 indicates the source and destination of an existing message path, and the dispatched message sequence number 172 indicates the sequence number of a dispatched message on an existing message path.

In the message path management table 170 in FIG. 20, the dispatched message sequence number 172 for the message path 171 "A → B" is "5, 8." This means, for example, that when receiving message path information from another node (step S61 in FIG. 19), message path information indicating that the dispatched sequence number for the message path "A → B" is "8" was received in addition to message path information indicating that the dispatched sequence number for the message path "A → B" is "5" (e.g., message path information 160 from node ID "2" shown in FIG. 18). If this message path management table 170 is the message path management table 35 of the leader node 10A, in step S64 in FIG. 19, the message processing unit 22 of the leader determines that the maximum value of the dispatched message sequence numbers for the message path "A → B" is "8." Therefore, in step S64, the message processing unit 22 of the reader adds a log entry of the "Message path added" type to the log (event log 34), with the existing message path 151 set to "A → B", the additional message path 152 set to "D → B", and the dispatched sequence number 153 set to "8".

The message path addition process in Figure 15 and the message path information reception process in Figure 19 described above have the following features:

If the new message path is the second message path for the message destination task, etc., a log entry of type "Message path added" is added to the log during the message path information reception process (step S64 in Figure 19), and consensus is reached between nodes on the addition of the message path. At this time, the dispatch of the message to the message destination is put on hold until consensus is reached on the addition of the new message path (step S52 in Figure 15).

Furthermore, once consensus on the message path addition is reached and the log entry is committed and applied, the message path is added to the message path management table 35 (step S55 in FIG. 15, step S61 in FIG. 19).

If the new message path is the first message path, dispatching of messages on the first message path is suspended until the message path is added to the message path management table 35.

If the new message path is the third or subsequent message path, messages to the message destination that occur after the new message path is added will be added to the log after the "Message path added" type log entry that corresponds to the addition of the new message path. As a result, dispatching will be put on hold until the addition of the new message path is complete.

(6-3) Improving message waiting time when adding a message path As mentioned above, when adding a second message path, dispatching of messages on the first message path is temporarily suspended until consensus is reached between nodes on the "message path added" type log entry and the message path is added to the message path management table. At this time, processing in the FT system 1 stagnates for the time that dispatching is suspended (message waiting time).

To prevent such processing stalls, one possible solution is to build the contents of the message path management table 35 to a certain extent during system testing, save the contents, and then restore them during actual operation. This method can reduce the frequency of dynamic message path additions to a certain extent, but it would be even better if it could be reduced even further.

Therefore, in this embodiment, the following two improvement methods can be adopted to further improve message waiting time.

(6-3-1) First Improvement Method In the first improvement method, when a node receives message path information regarding the addition of a second message path from another node, the node dispatches messages on the first message path up to the maximum value of the dispatched message sequence number written in the message path information, prior to adding the message path information to the message path management table 35 (step S61 in FIG. 19).

The first improvement method is suitable for situations where there is little increase or decrease in the number of nodes, and can minimize delays caused by deferred dispatching.

However, the first improvement method poses the following risk in situations where the number of nodes increases or decreases frequently. If a node that reports the maximum value M of the dispatched message sequence number in its message path information fails and stops before transmitting that message path information to the leader, the leader will add an "Add message path" type log entry to the log, with N < M as the dispatched message sequence number. If there is a node that has already dispatched up to M at this time, that node will not be able to apply the log entry as instructed by the leader and will need to be excluded from the cluster. Therefore, the more frequently the number of nodes increases or decreases, the greater the risk that the multiplexed system of nodes 10 in the FT system 1 will not be able to be maintained.

(6-3-2) Second Improvement Method In the second improvement method, in reaching a consensus on adding a message path, rather than reaching a consensus between nodes after the leader receives message path information regarding the addition of a second message path from all nodes as in the method described above, whenever the maximum value of the message completion sequence number is updated while the leader is receiving message path information regarding the addition of a second message path from all nodes, a consensus on adding a message path is reached between nodes based on a consensus protocol such as RAFT, and messages are dispatched until a consensus is reached. Below, the configuration and processing for realizing the second improvement method are described in detail.

When adopting the second improvement method, the FT system 1 introduces a new type of log entry, a "message path information" type log entry.

The purpose of the "Message Path Information" log entry is to reach an agreement among nodes when adding a second message path to a certain message destination that "a new message path will be added after dispatching at least N message sequence numbers on the existing message path." Since consensus can be reached with a response from a majority of nodes, by using this "Message Path Information" log entry, it is possible to preemptively dispatch message requests up to N before the leader receives message path information from all followers. The specific behavior when applying the "Message Path Information" log entry is to dispatch message requests that are stuck on an existing message path with a "blocked state" set, in order, up to the message request that corresponds to the dispatched message sequence number indicated in the "Message Path Information" log entry.

Figure 21 shows an example of a log entry of the message path information type. As shown in Figure 21, a log entry 180 of the message path information type is composed of a type 181 and message path information 182. Type 181 indicates the type of information contained in the log entry, including the node ID, existing message path, additional message path, and dispatched sequence number. Message path information 182 indicates the content of each type of information indicated in type 181.

Figure 22 is a sequence diagram showing an example of the processing procedure for adding a message path using the second improvement method. For the sake of explanation, Figure 22 shows a simplified node configuration with one leader (node ID = 1) and two followers (node IDs = 2 and 3). The processing of each step described below is performed by the message processing unit 32 of the corresponding node.

In FIG. 22, log entries 191 and 193 are examples of "message path information" type log entries added to the log (event log 34) in steps S71 and S73, and log entry 195 is an example of a "message path addition" type log entry added to the log (event log 34) in step S75. Furthermore, message path information 192 and 194 are examples of message path information broadcast in steps S72 and S74. Message path information 192 and 194 are notifications broadcast from one node to other nodes using a protocol specific to this embodiment that is not specified in RAFT.

As shown in Figure 22, first, node 1, the leader node, adds log entry 191 of its own node's "message path information" to the log (step S71). As a result of this addition to the log, a copy of the log is broadcast to follower nodes (nodes 2 and 3) according to the RAFT protocol. Each node then applies log entry 191 as soon as it has committed it. As a result, each node dispatches message requests that were blocked on message path "A → B" up to message sequence number "5."

Then, let us assume that follower node Node 3 broadcasts message path information 192 indicating that message sequence number "6" has been dispatched (Step S72).

In this case, node 1, which received message path information 192, notices that the dispatched message sequence number on message path "A → B" in message path information 192 is "6," which is greater than the dispatched message sequence number "5" indicated in log entry 191 added to the log in step S71. Therefore, node 1 adds log entry 193 of type "message path information" indicating the contents of message path information 192 to the log (step S73). This addition updates the dispatched message sequence number in the log entry of type "message path information" to "6." Then, just as with the addition to the log in step S71, node 1 broadcasts a copy of the log to follower nodes (nodes 2 and 3) in accordance with the RAFT protocol, and each node applies it as soon as the commit is complete.

Furthermore, suppose that follower node Node 2 subsequently broadcasts message path information 194 indicating that message sequence number "4" has been dispatched (Step S74). In this case, Node 1, which receives message path information 194, does not perform any special processing because the dispatched message sequence number on message path "A → B" in message path information 194 is "4," which is smaller than the maximum value of "6" that has been the largest number of dispatched messages up to that point.

When node 1, the leader node, receives message path information from all nodes, it generates a log entry of type "Message path addition" based on the log entry of type "Message path information" that was finally added to the log, and adds this to the log (event log 34) (step S75). Specifically, it adds to the log a log entry of type "Message path addition" 195 that sets the dispatched message sequence number on message path "A → B" to "6" and the added message path to "D → B". Note that the application processing after this log addition is the same as the processing described in Figure 12.

As described above, in the second improvement method, consensus is reached using the log entry of "message path information" when adding a message path, so the processing procedures for the message path addition process shown in Figure 15 and the message path information reception process shown in Figure 19 are partially modified as follows:

Figure 23 is a flowchart showing an example of the processing procedure for message path addition processing using the second improvement method. The flowchart in Figure 23 will be explained, focusing on the differences from the message path addition processing shown in Figure 15, and a description of the common processing will be omitted.

As shown in FIG. 23, after setting the first message path to a blocked state in step S52, the message processing unit 22 checks whether the node itself is the leader (step S81). If the node itself is the leader (YES in step S81), the message processing unit 22 adds a log entry of the node's "message path information" to the log (step S82). This step S82 process is realized by adding a log entry in the RAFT protocol. As explained in step S71 of FIG. 22, as a result of adding the entry to the log, a copy of the log is broadcast to followers in accordance with the RAFT protocol. On the other hand, if the node itself is not the leader, i.e., is a follower (NO in step S81), the message processing unit 22 broadcasts message path information of the existing message path to other nodes by message transmission specific to this embodiment, independent of the RAFT protocol, as explained in steps S72 and S74 of FIG. 22 (step S53).

Figure 24 is a flowchart showing an example of the processing procedure for message pass information reception processing using the second improvement method. The flowchart in Figure 24 will be explained, focusing on the differences from the message pass information reception processing shown in Figure 19, and a description of the common processing will be omitted.

As shown in FIG. 24, the message processing unit 22 of the node that received the message path information registers the message path information in the message path management table 35 (step S61), and if the node is the leader node (YES in step S62), it determines whether message path information regarding the addition of a new message path has been received from all other nodes (step S63).

Then, if the leader has not received message path information regarding the addition of a new message path from all other nodes in step S63 (NO in step S63), the message processing unit 22 of the leader checks whether the maximum value of the dispatched message sequence number has been updated by registering the message path information in the message path management table 35 in step S61 (step S91).

If the maximum value of the dispatched message sequence number has been updated in step S91 (YES in step S91), the message processing unit 22 of the leader adds a log entry of type "message path information" to the log based on the message path information received in step S61 (step S92), and terminates the message path information reception process. The processing of step S92 corresponds to the processing of step S71 in FIG. 22. In other words, if the "maximum value" of the dispatched message sequence number on an existing message path, including that of its own node, is updated among the message path information currently received from other nodes, the message processing unit 22 of the leader adds a log entry of type "message path information" to the log even if it is in the middle of receiving message path information from all nodes before receiving message path information.

On the other hand, if the maximum value of the dispatched message sequence number has not been updated in step S91 (NO in step S91), the message processing unit 22 of the leader terminates the message pass information reception process without performing any special processing. This processing procedure corresponds to the case where message pass information 194 is broadcast from the follower (node 2) in step S74 of Figure 22, and the leader (node 1) receives this message pass information 194 and does not perform any special processing.

To summarize the above explanation, the second improvement method for adding a message path has the following features when compared to the message path addition processing method described in (6-2):

The difference between the message path addition process in the second improvement method (6-2) is that a log entry of type "message path information" is first introduced. Then, instead of broadcasting the message path information to follower nodes, the leader node adds a log entry of type "message path information" to the log. As a result, a copy of the log is broadcast to follower nodes according to the RAFT protocol.

Also, when a leader node receives message path information from a follower node, if the maximum value of the dispatched message sequence number for an existing message path is updated, the leader node adds a log entry of type "Message Path Information" corresponding to the received message path to the log.

In addition, when a log entry of type "message path information" is committed and each node applies the contents of that log entry in response to this commit, message requests that are blocked on the existing message path are dispatched up to the dispatched message sequence number indicated in that log entry (message path information).

On the other hand, a common point between the second improvement method and the message path addition process (6-2) is that the follower broadcasts message path information to other nodes. The reason for this configuration is that if the leader stops due to a failure or other reason during the message path addition process and the follower who becomes the new leader continues the message path addition process, the information broadcast from each follower is required. Also, when the leader receives message path information from all nodes, it adds a log entry of type "Message path addition" to the log. Furthermore, the process for applying this "Message path addition" type log entry is also common to the message path addition process (6-2).

By incorporating the above-mentioned features, the second improvement method allows some of the blocked message requests to be safely dispatched before the leader has finished receiving message passing information from all nodes. "Safely" here means that even if some nodes leave during processing due to a failure or other reason, it will not result in a situation where synchronization cannot be maintained between nodes, as explained with reference to Figure 14. The second improvement method is particularly effective when the FT system 1 is configured such that there is a node that is slower to notify message passing information than other nodes and node departures are more likely to occur.

(7) Output processing unit 33
The processing performed by the output processing unit 33 will be described in detail below.

The output processing unit 33 receives external output requests as the processing results of each task 20, determines the content to be output externally in accordance with a predetermined policy, and transmits the determined content to the outside (terminal 3 such as a client).

As described above, in this embodiment, the task groups of each node 10 are guaranteed to behave deterministically. Therefore, when focusing on a particular task 20, essentially all nodes 10 will send the same external output requests in the same order.

On the other hand, there is no guarantee that the order of external output requests from different tasks 20 will match between nodes, and the order in which the external outputs are actually made is not deterministically determined (see Figure 25(A)). However, these external output contents are generally independent of each other, and it is assumed that the order of the external outputs does not affect the results of the system. If the order of external output requests from different tasks 20 is important, it is possible to match the output order between nodes by placing a common task (task C in Figure 25(B)) that relays the contents of each external output (see Figure 25(B)).

Figure 25 is an image diagram explaining the order guarantee of external outputs. In both Figures 25(A) and (B), external output requests from different tasks 20A and 20B (tasks A and B) are sent to the external output 39. The external output 39 can be interpreted as the output processing unit 33.

In the configuration of Figure 25 (A), task A sends message 101A requesting external output for "A → External #2" and "A → External #3" to external output 39, and task B sends message 101B requesting external output for "B → External #4" to external output 39. Here, for external output requests from the same task A, it is guaranteed that "A → External #3" will be output after "A → External #2." On the other hand, for external output requests from different tasks A and B, there is no guarantee that "A → External #2" or "B → External #4" will be output first, and the order in which they arrive at external output 39 (output processing unit 33) may differ between nodes.

The configuration in Figure 25(B) is the same as the configuration in Figure 25(A), but with the addition of a common task C that aggregates and relays the contents of each external output. As mentioned above, by having task C receive external output requests from different tasks A and B, the output order of external outputs from task C to external output 39 (output processing unit 33) can be made consistent between nodes.

The "predetermined policy" that the output processing unit 33 follows when determining the content to be output externally may be, for example, a first-come, first-served policy or a majority vote policy.

Under the first-come, first-served policy, the first external output request received from each replicated node (each node 10) is output externally, and the rest are discarded. Under the majority voting policy, a majority vote is held on external output requests that are thought to be identical and arrive from each replicated node (each node 10), the most likely content is selected and output externally, and the rest are discarded.

Furthermore, the FT system 1 of this embodiment may be provided with a dedicated external output device to output the processing results of the node 10 to an external device (e.g., terminal 3). However, by configuring it as shown in Figure 26 below, the external output of each node 10 can be adjusted and output to an external device even without a dedicated external output device.

Figure 26 is an illustration of a configuration for coordinating and outputting external outputs from multiple nodes. In Figure 26, when task A of each node sends an external output request, it broadcasts it to the external output 39 (output processing unit 33) of each node. The external output 39 (output processing unit 33) then compares the contents of the external output requests received from task A of each node, and transmits the one that is determined to be correct, for example by majority vote, to an external device (terminal 3). In the case of Figure 26, as shown in the enlarged view, the contents of the external output requests from node 1 and node 2 are "α", which is the majority, and therefore "α" is output externally.

Furthermore, in Figure 26, if an idempotent messaging protocol is used for communication between the external output 39 and the external device (terminal 3), duplicate messages can be discarded by the external device. For example, a sequence number can be assigned to the transmitted message on a session-by-session basis, and the messages can be retrieved one by one in sequence number order on the receiving side. In this way, even if a node fails and stops before an external transmission, the external device can receive the processing results without delay.

1 Fault-tolerant (FT) system 2 Communication network 3 Terminal 10 Node 10A Leader node (leader)
10B, 10C Follower node (follower)
20 Task 21 State (local state)
30 Event processing platform 31 Input processing unit 32 Message processing unit 33 Output processing unit 34 Event log 35 Message path management table 36 Message waiting queue 38 External input 39 External output

Claims (8)

A fault-tolerant system in which multiplexed nodes are connected via a communication network,
each of the plurality of nodes has a processor, a memory, and a storage area;
An application that runs by the processor in each of the nodes reading a program from the storage area into the memory and executing it is composed of one or more tasks that deterministically generate an output for an input,
one of the plurality of nodes is a leader node, and the nodes other than the leader node are follower nodes, and the leader node and the follower nodes are configured to perform the same task,
Each task is associated with a local state that indicates the state of the task itself,
The task's inputs include messages received from outside the system or from other tasks, and access to the local state associated with the task;
The outputs of the task include messages that the task sends to the outside of the system or to other tasks, and updates to the local state associated with the task;
a fault-tolerant system characterized in that, when the task sends a message to a destination task having a plurality of message senders, processors of the plurality of nodes reach an agreement among the plurality of nodes on a delivery order of the messages to the destination task, and deliver and process the messages in accordance with the agreed-upon delivery order. 2. The fault-tolerant system according to claim 1, wherein when the task sends a message to a destination task having a plurality of message senders, sending of the message to the destination task is temporarily suspended until an agreement is reached among the plurality of nodes regarding an order of message delivery to the destination task. 3. The fault-tolerant system according to claim 2, wherein when a new second message path is added from a third task different from one or more first tasks to a second task for which an existing first message path has been generated from the first tasks, the processor coordinates recognition among the plurality of nodes regarding the addition of the second message path. If the second message path to be added is a second message path for the second task,
The fault-tolerant system of claim 3, wherein the processor reaches a consensus among the plurality of nodes regarding up to which messages on the first message path should be dispatched before adding the second message path, and temporarily suspends transmission of messages on the first message path until such consensus is reached. When the second message path to be added is the second message path for the second task,
5. The fault-tolerant system of claim 4, wherein the processor of the leader node determines the maximum value of the sequence number of messages already dispatched on the first message path at each node, and reaches a consensus among the plurality of nodes to add the second message path after the message with the maximum sequence number has been dispatched on the first message path. When the processor of each of the nodes intends to add a second message path for the second task as the second message path, the processor of each of the nodes broadcasts message path information regarding the addition of the second message path to other nodes;
the message path information indicates a sequence number of a message that has been dispatched on the first message path at the node;
The fault-tolerant system described in claim 4, characterized in that the processor of the other node that receives the message path information dispatches messages on the first message path in its own node up to the maximum sequence number indicated in the message path information, prior to reaching an agreement among the multiple nodes as to how many messages should be dispatched on the first message path before adding the second message path. When the processor of the follower node intends to add a second message path for the second task as the second message path, the processor of the follower node broadcasts message path information regarding the addition of the second message path to other nodes;
5. The fault-tolerant system according to claim 4, wherein, while the leader node is receiving message path information regarding the addition of the second message path from all nodes, when the maximum value of the sequence numbers of messages already dispatched on the first message path indicated in the message path information received by the leader node from any node exceeds the maximum value of the sequence numbers of messages already dispatched on the first message path in the leader node, the nodes reach an agreement to dispatch messages on the first message path in their own nodes up to the maximum value of the sequence numbers indicated in the message path information. A data processing method using a fault-tolerant system in which multiplexed nodes are connected via a communication network, comprising:
each of the plurality of nodes has a processor, a memory, and a storage area;
An application that runs by the processor in each of the nodes reading a program from the storage area into the memory and executing it is composed of one or more tasks that deterministically generate an output for an input,
one of the plurality of nodes is a leader node, and the nodes other than the leader node are follower nodes, and the leader node and the follower nodes are configured to perform the same task,
Each task is associated with a local state that indicates the state of the task itself,
The task's inputs include messages received from outside the system or from other tasks, and access to the local state associated with the task;
The outputs of the task include messages that the task sends to the outside of the system or to other tasks, and updates to the local state associated with the task;
a data processing method characterized in that, when the task sends a message to a destination task having a plurality of message senders, processors of the plurality of nodes reach an agreement among the plurality of nodes on a delivery order of the messages to the destination task, and transmit the messages in accordance with the agreed-upon delivery order.