Message passing among interdependent parallel processes using a shared memory

The field relates generally to inter-process communication among a plurality of interdependent parallel processes.

Many compute sites support a combination of enterprise and High Performance Computing (HPC) applications. HPC workloads tend to be comprised of a plurality of interdependent parallel processes that must run synchronously in time due to the large amount of inter-process communication. Typical HPC workloads execute on clusters of compute nodes interconnected by a low-latency network. Enterprise workloads, on the other hand, tend to be independent tasks that do not require a significant amount of inter-process communication.

Unfortunately, the compute infrastructure historically employed for an enterprise site does not closely resemble the compute infrastructure historically employed for HPC sites. Therefore, distinct compute sites have typically been employed for each workload type. It is anticipated, however, that HPC and enterprise computing systems will converge. For typical compute systems, there is significant pressure to support a variety of services using a shared converged infrastructure.

In order for HPC applications to be successfully deployed in an enterprise compute infrastructure, however, a need exists for improved techniques for inter-process communication among the plurality of interdependent parallel processes. A further need exists for inter-process communication among the plurality of interdependent parallel processes using a shared memory in an enterprise environment.

Illustrative embodiments of the present invention provide for message passing among a plurality of interdependent parallel processes using a shared memory. In one embodiment, inter-process communication among a plurality of interdependent processes executing on a plurality of compute nodes is performed by obtaining a message from a first process for a second process; and storing the message in a memory location of a Peripheral Component Interconnect Express (PCIE)-linked storage device, wherein the second process reads the memory location to obtain the message.

In at least one exemplary embodiment, the memory location comprises a unique value for the first process and the second process. The memory location is optionally locked until the second process reads the memory location to obtain the message. For larger messages, the memory location optionally stores an offset to a second memory location where at least a portion of the message is stored.

In a further exemplary embodiment, the message is persistently stored in the PCIE-linked storage device for an asynchronous checkpoint until the message is no longer required for an asynchronous restart.

The illustrative embodiments described herein provide significant improvements relative to conventional arrangements. In some of these embodiments, the duplication of conventional message logging schemes that must send and store messages is removed by sending messages via a shared storage medium.

FIG. 1 illustrates a plurality of interdependent parallel processes executing on a plurality of compute nodes and communicating by means of a shared memory in accordance with aspects of the invention;

FIG. 2 illustrates message passing among two exemplary processes;

FIG. 3 illustrates exemplary pseudo code for MPI-Send and MPI-Receive functions incorporating aspects of the present invention; and

FIGS. 4A and 4B illustrate exemplary pseudo code for helper functions for the MPI-Send and MPI-Receive functions of FIG. 3 .

Aspects of the present invention provide improved techniques for inter-process communication among a plurality of interdependent parallel processes executing on a plurality of compute nodes. One aspect of the invention provides Message Passing Interface (MPI) communications among interdependent parallel processes using a shared memory in an enterprise environment. While exemplary embodiments of the present invention employ low-latency MPI communications and the associated MPI application programming interface (API) and library, other message passing paradigms can be employed, as would be apparent to a person of ordinary skill in the art.

In addition, while exemplary embodiments of the present invention employ Peripheral Component Interconnect Express (PCIE) communications, other low latency communications can be employed, as would be apparent to a person of ordinary skill in the art. For example, PCIE-linked flash storage devices, such as DSSD™ flash storage devices from EMC Corp., provide a latency of message passing of approximately 50 microseconds, which may be lower than general Ethernet latency of around 100 microseconds (which may be even slower for MPI message passing over Ethernet).

Generally, as used herein, low-latency communications employ a single socket, while high-latency communications employ multiple sockets. PCIe is a communication network within a single socket and is therefore low latency. DSSD™ flash storage devices therefore create a cross socket communications channel that operates within socket latencies.

One aspect of the invention employs an enterprise infrastructure with one or more PCIE-linked flash storage devices, such as DSSD™ flash storage devices, to implement an MPI message passing API and library such that HPC workloads can run efficiently on an enterprise architecture.

In this manner, the messages are stored persistently as they are passed through the exemplary PCIE-linked storage device. Thus, according to another aspect of the invention, the message logging of the present invention permits asynchronous checkpointing. In this way, each process can selectively checkpoint itself independently from other processes. Asynchronous checkpointing normally requires that the messages passed between the asynchronous checkpoints must be stored persistently. The disclosed message passing implementation via the exemplary PCIE-linked storage device, however, already implicitly persistently stores passed messages. To enable this, applications must merely not immediately delete messages upon their receipt but delay the removal until the messages are no longer needed for asynchronous restart.

Additional aspects of the invention extend existing MPI libraries with a new communications paradigm, thereby augmenting their existing support for traditional communication mechanisms, such as Infiniband and Ethernet, with an implementation that uses PCIE-linked storage devices for message passing. For example, DSSD™ flash storage devices are an example of PCIE-linked flash storage devices. Exemplary pseudo code for the extended MPI libraries to implement aspects of the invention is discussed below in conjunction with FIGS. 3 through 5 .

FIG. 1 illustrates a plurality of interdependent parallel processes P 1-1 through P M-N M executing on a plurality of compute nodes 110 - 1 through 110 -M. It is noted that the first index for each process indicates the compute node 110 upon which the process executes and the second index for each process indicates the rank of the process on the respective compute node 110 . The number of processes executing on each compute node 110 can vary. Each compute node 110 comprises a PCIE bus 130 - 1 through 130 -M. The interdependent parallel processes communicate with one another using the respective PCIE bus 130 and a PCIE switch 150 , as discussed further below.

One or more PCIE-linked storage devices 160 , such as a PCIE-linked DSSD™ flash storage device from EMC Corp., are connected to the PCIE switch 150 by means of a PCIE bus 170 .

FIG. 2 illustrates message passing 200 among two exemplary processes 0 and 1 over the PCIE switch 150 of FIG. 1 . In the exemplary embodiment of FIG. 2 , process 0 sends data and process 1 receives data, making the exchange of data cooperative. Data must both be explicitly sent and received by the respective process. A change in the memory of the receiver process 1 is made with the participation of the receiver process 1.

In accordance with aspects of the present invention, synchronous messaging is implemented by creating a memory offset in the exemplary PCIE-linked storage device 160 using a (sender_id, receiver_id) pair. For small messages, the messages can be stored directly at the location identified by the memory offset. Larger messages can optionally employ an indirection in which the target location stores an offset to the larger message.

For asynchronous messaging, each receiver can have a mailbox address into which asynchronous messages can be stored. Thus, there is no need for a caller to hold the memory copy or for the MPI library to copy and buffer the message. Instead, since the message passing implementation is done by storing the message into the exemplary PCIE-linked storage device 160 , there is no need for the caller, or the MPI library on behalf of the caller, to store the message.

Local Area Multicomputer (LAM) is an exemplary MPI programming environment and development system for heterogeneous computers on a network. LAM/MPI is an existing implementation of the MPI communication standard. LAM/MPI allows a plurality of compute nodes forming a computer cluster to act as a single parallel computing resource. For a more detailed discussion of LAM, see, for example, http://yclept.ucdavis.edu/Beowulf/notes/man-lam/lam.man.html, incorporated by reference herein.

As noted above, aspects of the invention extend existing MPI libraries, such as LAM, to provide an implementation that uses shared memory in the form of PCIE-linked storage devices for message passing. In particular, aspects of the invention provide new implementations for MPI-Send and MPI-Receive functions in the exemplary LAM/MPI implementation. The exemplary MPI-Send function writes a message to the exemplary PCIE-linked storage device 160 and the exemplary MPI-Receive function reads a stored message from the exemplary PCIE-linked storage device 160 .

FIG. 3 illustrates exemplary pseudo code 300 for MPI-Send and MPI-Receive functions incorporating aspects of the present invention. The exemplary pseudo code 300 allows existing (i.e., unmodified) applications to call MPI-Send and MPI-Receive functions as in the past, and the internal library implementations for MPI_Send and MPI_Receive that are trigged by the function calls are shown in FIG. 3 .

As shown in FIG. 3 , the exemplary MPI_Send function determines the rank of the sending process using a my_rank process, locks the allocated mailbox and places the message in a buffer associated with the allocated mailbox on the exemplary PCIE-linked storage device 160 . The exemplary MPI_Receive function determines the rank of the receiving process using the my_rank process and tries to receive the stored message until it appears (in case the receiver tries to receive the message before it is sent). In addition, the exemplary MPI_Receive function locks the allocated mailbox and reads the message from the allocated mailbox on the exemplary PCIE-linked storage device 160 and copies it in a buffer back to the receiving process.

FIGS. 4A and 4B illustrate exemplary pseudo code 400 and 450 , respectively, for helper functions for the MPI-Send and MPI-Receive functions of FIG. 3 . As shown in FIG. 4A , the exemplary pseudo code 400 comprises a copy_message helper function 410 to read a message from a mailbox that is called by the MPI-Receive function of FIG. 3 .

In addition, the exemplary pseudo code 400 comprises a place_message helper function 420 to put a message into a mailbox that is called by the MPI-Send function of FIG. 3 . The exemplary pseudo code 400 also comprises an unlock_mailbox helper function 430 to unlock the mailbox when the message passing is complete.

As shown in FIG. 4B , the exemplary pseudo code 450 comprises a lock_mailbox helper function 460 to ensure that the sender and receiver processes have mutual access to the virtual mailbox.

In addition, the exemplary pseudo code 450 comprises a lock_address helper function 470 to obtain the lock address. The exemplary pseudo code 450 also comprises a mailbox_address helper function 480 to obtain the mailbox address using different addresses to ensure no mailbox conflicts. Thus, the mailbox_address helper function 480 obtains a unique value for each sender/receiver pair.

Among other benefits, the passing of messages through the PCIE switch 150 instead of the network switch (not shown in FIG. 1 ), reduces the bandwidth required for a low-latency network. In addition, while previous message logging schemes have duplicated the messages in order to send them as well as to store them, aspects of the present invention remove the duplication by sending them via the storage medium.

In addition, parallel program resilience to failure has traditionally been achieved through an expensive state dump mechanism in which the parallel program imposes an expensive barrier or fence to quiesce the network and then using the storage system in an extremely inefficient bulk synchronous large data transfer. Alternatives to this checkpoint/restart resilience method include message logging which is expensive both in terms of latency and capacity costs. The message logging done heretofore duplicates the messaging by both sending it through the network as well as storing the message in a persistent storage. By doing message passing natively through the persistent storage in accordance with aspects of the invention, the duplication is removed.

The compute nodes 110 - 1 through 110 -M (or portions thereof), comprising a processor, memory and network interface components as described above, are examples of what is more generally referred to herein as a “processing device.” Each of the devices herein may similarly be implemented as a processing device comprising a processor, memory and network interface components. Thus, one or more of the exemplary compute nodes 110 - 1 through 110 -M in the exemplary enterprise environment comprises a processor coupled to a memory. The processor may comprise a microprocessor, a microcontroller, an application-specific integrated circuit (ASIC), a field-programmable gate array (FPGA) or other type of processing circuitry, as well as portions or combinations of such circuitry elements. The memory may comprise random access memory (RAM), read-only memory (ROM) or other types of memory, in any combination. The memory and other memories disclosed herein may be viewed as examples of what are more generally referred to as “computer program products” storing executable computer program code. Also optionally included in one or more of the exemplary compute nodes 110 is network interface circuitry. The network interface circuitry allows the exemplary compute nodes 110 to communicate over a network (not shown in FIG. 1 ) with other devices. The network interface circuitry may comprise, for example, one or more conventional transceivers.

It is to be appreciated that the particular operations and associated messaging illustrated herein are exemplary only, and numerous other types of operations and messaging may be used in other embodiments.

It should therefore be understood that in other embodiments different arrangements of additional or alternative elements may be used. At least a subset of these elements may be collectively implemented on a common processing platform, or each such element may be implemented on a separate processing platform.

Also, numerous other arrangements of computers, servers, storage devices or other components are possible in the exemplary enterprise environment of the present invention. Such components can communicate with other elements of the exemplary enterprise environment over any type of network or other communication media.

As indicated previously, components of an exemplary enterprise environment as disclosed herein can be implemented at least in part in the form of one or more software programs stored in memory and executed by a processor of a processing device. A memory having such program code embodied therein is an example of what is more generally referred to herein as a “computer program product.”

The exemplary enterprise environment or portions thereof may be implemented using one or more processing platforms each comprising a plurality of processing devices. Each such processing device may comprise processor, memory and/or network interface components.

As indicated above, message passing using shared memories such as that described herein can be implemented at least in part in the form of one or more software programs stored in memory and executed by a processor of a processing device such as a computer or server. A memory or other storage device having such program code embodied therein is an example of what is more generally referred to herein as a “computer program product.” Certain system components are implemented using a combination of software and hardware.

It should again be emphasized that the above-described embodiments of the invention are presented for purposes of illustration only. Many variations and other alternative embodiments may be used. For example, the disclosed techniques are applicable to a wide variety of other types and arrangements of compute nodes and interrelated processes that can benefit from the message passing functionality as described herein. Also, the particular configurations of system and device elements shown in FIG. 1 can be varied in other embodiments. Moreover, the various assumptions made above in the course of describing the illustrative embodiments should also be viewed as exemplary rather than as requirements or limitations of the invention. Numerous other alternative embodiments within the scope of the appended claims will be readily apparent to those skilled in the art.