Patent ID: 12197290

DETAILED DESCRIPTION

The present disclosure will now be described with reference to the figures, which in general relate to enabling lockstep processing using a processing environment having multiple processing nodes. The technology allows for a fault tolerant processing environment to be created wherein multiple processors can be configured as worker nodes and redundant nodes, with a failed worker node replaced programmatically by a manager node. Each of the processing nodes may include a processor and memory associated with the processor and communicate with other processing nodes using a network. The manager node crates a message passing interface (MPI) communication group having worker nodes and redundant nodes, instructs the worker nodes to perform lockstep processing of tasks for an application, and monitors execution of the tasks. If one of the processing nodes returns faulty results (or is otherwise determined to have faults such as communication faults), the manager node creates a replacement worker node from one of the redundant processing nodes and creates a new communications group. It then writes check-back data to the nodes in the new communication group (including the replacement worker node) and instructs those nodes in the new communications group to resume processing based on the state of the checkpoint backup data.

The technology allows for the use of fault detection based on task outputs, as well as MPI fault tolerance, to positively detect the failed node and recover provide a new worker using MPI spawn. Therefore, the Hadoop configured new processing node and connections can be recovered and managed initialized from MPI communicators. Hadoop can establish TCP connections between the processing nodes to allow them to communicate with each other, avoiding many disadvantages of the methods that using only MPI or MapReduce for fault tolerance.

The design uses storage to maintain the checkpoint backup data and copy the data to any new processing nodes when needed. It can effectively prevent the data from damage when a processing node fails. It also makes the selection of new processing nodes flexible because selection is not impacted by the location of backup data. The technology thus provides a fault tolerant distributed computing environment which recovers from processing failures.

FIG.1is a block diagram of a distributed processing environment100. The distributed processing environment100ofFIG.1includes a plurality of processing nodes (N1-N8, . . . Nn), each node having, for example, a hardware configuration illustrated inFIG.2, although it will be understood that the technology is not limited to a particular hardware configuration for each processing node. The distributed processing environment100is configured and managed using message passing interface (MPI), Hadoop and MapReduce. Hadoop is software library and framework that facilitates using a network of many computers. Hadoop includes a storage component, the Hadoop Distributed File System (HDFS), and the MapReduce programming model.FIG.1illustrates a number of nodes N1-Nn, in the processing environment some of which may comprise processing elements (PE-0-PE-n, PE1-1, PE1-2). The MapReduce-MPI manager190ofFIG.1will employ a number of nodes “C” comprising a Hadoop processing cluster150which are used to execute an application by executing tasks of the application. A number (R) of redundant nodes115may be used for backup of any worker node in case a failure happens. As illustrated in the simplified example inFIGS.1, C=5 and R=2. The distributed processing environment100may also include other nodes indicated at N7-N8which are not in the communicator group220. Each of the nodes in the domain may be connected by a network path125, such as a bus, network cable or other interconnection. A MapReduce program may be composed of a Map procedure (method) that performs filtering and sorting and a Reduce method that performs a summary operation. The “MapReduce System” (also called “infrastructure” or “framework”) orchestrates the processing by marshalling the distributed nodes, running the various tasks in parallel (and in this technology, in lockstep), managing all communications and data transfers between the various parts of the system, and providing for redundancy and fault tolerance.

In various embodiments, the processing nodes (N1-N8, . . . Nn), may be fabricated on a common semiconductor substrate and packaged accordingly. In other embodiments, the processing nodes may be coupled to a common circuit board or multiple circuit boards which are coupled by bus125, which in such embodiment may comprise a network cable. As illustrated inFIG.1, an advantageous application of the present technology is for use in a motor vehicle for controlling features of the motor vehicle. Given the need for safety in vehicle operations, a fault tolerant system such as that disclosed herein is useful in vehicle applications.

FIG.2illustrates an exemplary hardware configuration of the processing nodes illustrated inFIG.1. A processing node configured as a processing element (PE)200may comprise any of the nodes used as processing elements described herein. As shown inFIG.2, each processing element200may comprise a memory210, cache220, Central Processing Unit (CPU)230and network interface240. The CPU230may comprise any type of electronic data processor. Memory210may comprise volatile memory and a non-volatile memory. The memory210may comprise any type of storage device configured to store data, programs, and other information and to make the data, programs, and other information accessible via the bus125. The memory210may comprise any type of system memory such as static random-access memory (SRAM), dynamic random-access memory (DRAM), synchronous DRAM (SDRAM), read-only memory (ROM), a combination thereof, or the like. In an embodiment, the memory210may include ROM for use at boot-up, and DRAM for program and data storage for use while executing programs. Cache220is utilized to increase the performance of each PE by loading needed data from memory for rapid access by CPU230. Network interface140couples PE200to the network bus125.

In the above description, the CPU is used interchangeably with the term processor. However, a processor contains other discrete parts within it, such as one or more memory caches for instructions and data, instruction decoders, and various types of execution units for performing arithmetic or logical operations. A processor may be understood to include a multiprocessor system with more than one CPU per processing node, as well as a multicore CPU with multiple execution cores on one CPU.

The memory210may include computer-readable non-transitory media which stored instructions to enable the CPU to perform the methods described herein. Specific processing nodes may utilize all of the components shown, or only a subset of the components, and levels of integration may vary from device to device. Furthermore, each processing node may contain multiple instances of a component, such as multiple processing units, processors, memories, etc.

FIG.3illustrates some of the processing nodes ofFIG.1configured as a processing cluster, such as cluster150ofFIG.1, to perform fault tolerant lockstep processing. Node PE-0is configured a master node, while nodes PE-1, PE-2. . . PE-n represent worker nodes. In the example ofFIG.3, three (3) worker nodes are shown. It should be understood that any number of worker nodes (n) may be utilized in a processing cluster. Master node PE-0configures the cluster, manages application tasks, and detects failures of worker nodes by monitoring the status of each worker node. Once a failure on a worker node is detected, the master node will create a replacement worker node from the pool of redundant worker nodes (such as pool115ofFIG.1) and reschedule the affected tasks to the replacement worker node. The failed worker node is then removed from the processing cluster, but the total number of worker nodes remains constant and processing redundancy within the cluster150is maintained.

MPI allows communications between all processes in a communicator at once to distribute data or perform a data reduction. As illustrated inFIG.1, the nodes in cluster150as well as reserve resources115are part of a communicator group130. Each node in the communicator group130, including the redundant nodes115, has an MPI agent. Each node in the cluster130is numbered and identified using the MPI agents.FIG.3illustrates MPI agents316-1,316-2, . . .316-nin the worker nodes which communicate with each other and the master node via network bus125. In one embodiment, a TCP/IP network may be enabled on bus125to allow the processing elements to communicate via TCP/IP.

Manager node PE-0includes a lockstep controller302, MPI manager304, Hadoop manager306and its own MPI agent308. The MPI manager304works with the lockstep controller302to execute and monitor the tasks, and the Hadoop manager306to define the computing cluster130. The MPI manager304and controller302are responsible for distributing tasks to each of the PEs. The lockstep controller302may include a job tracker, not shown inFIG.3, which allows the manager node to distribute and keep track of lockstep processing on each of the worker nodes PE-1, PE-2. . . PE-n. Each worker node PE-1, PE-2. . . PE-n includes a task tracker314-1,314-3, . . .314-ntracking respective tasks312-1,312-3, . . .312-n. HDFS allows for the storage of large amounts of data across multiple physical memories. HDFS achieves reliability by replicating the data across multiple nodes. MapReduce-MPI used with the HDFS duplicates data so that nodes may use local storage provided directly to a processor in a node. MapReduce takes advantage of local storage to avoid network bottlenecks when working with large data. As such, an HDFS320is distributed across multiple data nodes318-1,318-2. . .318-nwhich are associated with each worker node PE-1, PE-2. . . PE-n. Each data node318-1,318-2. . .318-nserves up blocks of data over the network bus125using a block protocol specific to HDFS. The file system may utilize TCP/IP sockets for communication via a network bus125. Each worker node and data node may use remote procedure calls (RPC) to communicate between each other.

FIGS.4and5A-5Billustrate the replacement of a failed PE in accordance with the present technology. In accordance with the technology, each worker node PE-1, PE-2. . . PE-n processes tasks assigned to it by the lockstep controller. When a failure is detected in the output of a PE, in this case PEn inFIG.4, the manager node uses MPI_Comm_Spawn to create a new worker node PE-N+1 When a worker dies, MPI_Comm_spawn replaces it, and continues processing with no fewer workers than before. The processes of a parallel worker may communicate with each other using an ordinary intercommunicator. For parallel workers, this is a simpler way to form the communicators group. Fault tolerance in this situation resides in the overall master-slave structure.

FIGS.5A and5Billustrate provision of data from the original data nodes comprising the HDFS file system to a data node318-4for a replacement worker node. As illustrated inFIG.5A, each task tracker312-1,312-2, . . .312-nprovides task results to a job tracker502in the lockstep controller302. The job tracker502thus maintains the state of each of the outputs of the processing elements of the worker nodes in the communications group. In the example ofFIG.5A, job tracker502receives three results and the lockstep controller determines that the results from data node318-3indicate a failure. As illustrated inFIG.5B, the state of node318-3represented by tasks510-3and514-3are transferred by the lockstep controller to a new, replacement data node318-4. In addition, checkpoint backup data504is provided to the data node318-4. By integrating with the fault tolerant mechanisms of MPI, the method can supply new worker PEs to the system after failed PEs are removed. Therefore, the consistency of the system throughput is maintained after failure recovery. The method can also recover communications between the PEs without significantly impacting the tasks executed on the system.

MPI uses checkpoints to back up the task status for fault tolerance, which takes periodic system snapshots and stores application status information in persistent storage units. If a failure occurs, the most recent status information can be retrieved and the system recovered from the checkpoints. Checkpoints are saved in a persistent storage unit that is not affected by the failure of a single computing element. InFIG.5B, the MPI checkpointing mechanism forms the checkpoints on the master node PE-0as check point data504and writes the data needed to restart the application. The load can then be rescheduled for processing amongst the same number of processing elements (three in this example) as existed prior to the failure of the worker processing element

FIG.6is a flowchart illustrating a method in accordance with the technology for fault tolerant lockstep processing. At610, MPI initializes a communicator group. In one embodiment, the manager processing element uses MPI to initialize the communicator group with MPI_COMM_WORLD. A communicator group represents a processing environment in an ordered set of processing environments initialized by MPI. The manager PE collects information from all communicators (MPI agents) from the distributed system. At620, Hadoop configures processing cluster130and reserved resources115. Hadoop initialization may comprise providing the named PE, and job trackers to each of the processing nodes in MPI_Comm_World. At625, lockstep processing of an application begins by the lockstep controller302distributing tasks to each of the worker node processing elements in the processing cluster150in parallel. This includes sending the node numbers, node names and the HDFS data path in which the application will run along with application file names and instructions. At630, the method checks to determine whether or not a processing element failure has occurred by checking the task ouput data is returned to the lockstep manager302. Additional details on checking for failures is discussed below with respect toFIG.8. If no failure is detected at630, the lockstep processing continues with all worker node PEs in the cluster contributing their results. If a worker node PE failure is detected at630, then the system moves to replace the failed PE with the new PE at640. Step640may be broken down into sub steps650,660,670,680, and690.

At650, a new communicator group is initialized using MPI_Comm_Spawn which provides New_MPI_COMM_WORLD with the same size as the original MPI_COMM_WORLD. At step660, the lockstep manager outputs a new communicator information to the map reduce master PE (PE-0in the previous FIGS.), name processing element and job trackers. This includes providing individual threads and processes of each thread to reflect the PE state in the new worker PE. At670, the manager node PE-0(the map reduce master processing element, name processing element and job trackers) reconfigure the processing environment by providing state information to the new worker node. At680, data from memory and storage is provided to the newly spawned worker processing element.

FIG.7is a flowchart illustrating a communicator group initialization procedure. At730, the total communicator number=C is set at master node PE-0. At740, the communicator group is initialized with n=C+R communicators using MPI_INIT and MPI_COMM_WORLD, where R is the reserved number of communicators. Information from all communicators is collected at750and an established state is updated after all the information is collected at760. At770, a normal operating mode is entered allowing each node to communicate via MPI.

Two forms of errors may be detected at step430—application errors and native MPI error detection.

FIG.8illustrates a method whereby an application result is determined to be in error which may indicate a failure in a processing node. For each application job at810, the job tracker802submits the job to all PEs in the communication group130along the data location in HDFS. As the Pes perform the job, the job tracker802receives the PEs job results at830. At840, a determination is made as to whether there is an error in the results between different PEs. There may be any number of reasons for a disparity in the results of a PE. Transient faults may be due to electrical noise or external radiations. Radiation can, directly or indirectly, induce localized ionization events capable of upsetting internal data states. While a physical instance may cause a data error, it is possible the computing circuit itself is undamaged and the system experiences a transient fault. However, multiple faults can indicate a failed processing node and hence a permanent fault in the system. To determine whether one of the PEs in the cluster has an error, one may utilize majority voting—the results of the majority of each of the PEs in the cluster are taken as valid. If no error is present at840, the method moves to the next task at860. If a failed PE is detected at840, then at850, a replacement PE process begins as described at640above.

A second type of failure may be detected by MPI. MPI has a highly accurate and complete failure detector. An application may be notified of a process failure when it attempts to communicate directly or indirectly with the failed process through the return code of the function and the error handler set on the associated communicator. The designed MPI mechanism realizes a recover-and-continue solution that re-spawns only failed processes if an error occurs while keeping living processes within their original processors or nodes.

To minimize the impact of the failure recovery process to the system, in one embodiment, a component, such as an MPI error handler, periodically checks the status of the MPI communicators. If a communicator failure is detected, the MPI will try to re-connect.

FIG.9illustrates the method by which the intermediate results of the same processes on different nodes are checked. By comparing the intermediate results of each worker node for a given task (TASK A, in this example), one can determine whether each task or process has an error on one of the worker nodes. Illustrated inFIG.9are three worker nodes PE-1, PE-2and PE-3, and one master node PE-0along with two redundant nodes PE4and PE5. For each task, the MPI master node sets different test points along the time sequence of each process or may set different test points based on MPI's calculation of the task's output characteristics. At each test point, for example point925, each processing environment submits its own intermediate data (Synch) to the MASTER PE (PE-0) in the communication group130. The MASTER PE receives the job results of each PE and determines whether the results between different PEs are wrong by a vote. At point925, all test results are the same. The MPI Master Lockstep controller902may include components such as ab MPI error processor and MPE communicator, with the MPI error processor regularly checking the calculation results of the MPI communicator. If an abnormal calculation result is detected, such as a minority in the vote, as indicated from PE-3at time935, the MPI Master Lockstep Controller902may continue to monitor additional test points or, as indicated inFIG.9, may begin the process or replacing PE-3. If this error occurs repeatedly on the same node, MPI will use this node as the error node, shut down the node, and find a new node (PE-4in this example) to replace it. The present technology thus uses the MPI mechanism to implement a PE recovery and continue solution that regenerates the failed process only when an error occurs, while keeping the active process in its original processor or node.

FIG.10illustrates a variation of the present technology whereby 10 nodes are split into three different work groups, all performing parallel computing on the same task (TASK A). Again, the MPI master node sets different test points along the time sequence of each process. The 10 nodes (PE-0-PE-9) are divided into three work groups for distributed processing of a task. As each work group performs distributed computing, the MPI master node902will distinguish whether each PE in each group runs the same work. G1P1and G2P1represent PE1of Group1 and PE1of Group2, respectively. These two PEs are in different groups but have the same role in parallel computing. The calculations of these two nodes should be consistent, and their results can be compared.

Therefore, the present technology utilizes a process and character comparison table in main PE-0. The structure of the table is as follows:

MPI workgroup (G)111222333Logical PE number123123123Physical PE number (PE-n)123456789

In this table, G1P1and G2P1correspond to physical devices PE1and PE4. Similarly, G3P1corresponds to physical device PE7. By comparing their intermediate results on a given task, one can determine whether each process has an error. As inFIG.9, for each application job (TASK A) at 10 locations, the MPI master node sets different test points (Proc 1, Proc3, Proc3) along the time sequence of each process, or may set different test points based on MPI's calculation of task output characteristics.

In the example ofFIG.10, each process submits its own intermediate data to the MASTER PE in the communication group130. When performing a job, MPIMASTER PE receives the job results of each PE and determines whether the results between different PEs are wrong by vote. Components such as the MPI error processor regularly check the calculation results of the MPI communicator. If an abnormal calculation result is detected, such as a minority in the vote, MPI will continue to monitor. If this error occurs repeatedly on the same node, MPI will use this node as the error node, shut down the node, and find a new node to replace it.

FIG.11shows a scenario wherein each different processing elements use different algorithms to solve a problem. For example, different AI algorithms may be used to identify an object and their results will be sent to a lockstep controller1102for a majority vote on the algorithm results from each PE. Since different algorithms performed on each different PE, the calculation methodologies adopted by each PE are different, and as such, their intermediate results are not comparable. Lockstep controller1102can only compare each PE's final output after the task is completed. On the other hand, different PEs should complete their work within a time frame defined by the system as a whole. For example, in a self-driving automotive application, the time consumption of each PE to identify an object should be short enough to allow control systems in the vehicle to react to the object if needed. The MPI master lockstep controller PE-01102sets a test point, such as at time1110, along the time sequence of each process. The time of the test point should before the reaction time minimum for a car in the above scenario, but long enough to complete majority voting of the results.

For a response time is defined as T, which is determined by the vehicle speed, road conditions and the driver, and/or configuration by the driver, and a slowest vote processing time is defined as V, which is determined according to the empirical value of the actual system, the maximum algorithm processing time T(comp) should satisfy T(comp)<T-V. If the algorithm processing time T(comp) is reached but a PE cannot send the results to lockstep controller1102, the lockstep controller1102judges that the PE is failed, and removes it from the current vote. The lockstep controller1102receives the job results of other PEs and determines whether the results between different PEs are wrong by vote. The final result output is the vote of all PEs.

FIG.12is a flowchart illustrating a specific version of the method illustrated inFIG.6. At1210, manager processing element uses MPI to initialize the communicator group with MPI_COMM_WORLD. It establishes connections with each PE, building a communication group that includes all communicators, and ensures that they are up and available.

At1220, the manager node sends the node numbers, names, and folder path in HDFS where the MPI process will run, along with application instructions. At1220, a remote procedure call (RPC) is used to initialize the independent, TCP connections with each PE. At1230MPI_Comm_Spawn is used to create communicators. For parallel workers, this is a simpler way to form the communicator group. When a worker dies, MPI_Comm_spawn replaces it, and continues processing with no fewer workers than before. At1235, each node performs execution of application tasks in lockstep and returns the results to the lockstep controller. Each node can deploy multiple threads. A node is assumed to be failed if the status is down. MPI implementation returns an error code if a communication failure occurs such as an aborted process or failed network link.

At1240, MPI Init establishes connection with each PE to build a new communicator group NEW_MPI COMM_WORKLD. At1250, sends size N node numbers, node names, a folder path on which the MPI process will run, and the file names with application instructions.

At1260, the map reduce Master Node, name PE and Job trackers reconfigure the communicator group. At1270, the MPI process1200sends size n node numbers, node names, the folder path on which the MPI process will run, and the file names with application instructions. RPC re-initializes independent, long-term TCP connections.

At1270, check point data is loaded from storage. At1280, MPI loads the checkpoints from storage, and at1280, the Hadoop nodes perform parallel execution independent of MPI applications.

For purposes of this document, it should be noted that the dimensions of the various features depicted in the figures may not necessarily be drawn to scale.

For purposes of this document, reference in the specification to “an embodiment,” “one embodiment,” “some embodiments,” or “another embodiment” may be used to describe different embodiments or the same embodiment.

For purposes of this document, a connection may be a direct connection or an indirect connection (e.g., via one or more other parts). In some cases, when an element is referred to as being connected or coupled to another element, the element may be directly connected to the other element or indirectly connected to the other element via intervening elements. When an element is referred to as being directly connected to another element, then there are no intervening elements between the element and the other element. Two devices are “in communication” if they are directly or indirectly connected so that they can communicate electronic signals between them.

Although the present disclosure has been described with reference to specific features and embodiments thereof, it is evident that various modifications and combinations can be made thereto without departing from scope of the disclosure. The specification and drawings are, accordingly, to be regarded simply as an illustration of the disclosure as defined by the appended claims, and are contemplated to cover any and all modifications, variations, combinations or equivalents that fall within the scope of the present disclosure.

The foregoing detailed description has been presented for purposes of illustration and description. It is not intended to be exhaustive or to limit the subject matter claimed herein to the precise form(s) disclosed. Many modifications and variations are possible in light of the above teachings. The described embodiments were chosen in order to best explain the principles of the disclosed technology and its practical application to thereby enable others skilled in the art to best utilize the technology in various embodiments and with various modifications as are suited to the particular use contemplated. It is intended that the scope be defined by the claims appended hereto.