In-node aggregation and disaggregation of MPI alltoall and alltoallv collectives

Abstract

An MPI collective operation carried out in a fabric of network elements by transmitting MPI messages from all the initiator processes in an initiator node to designated ones of the responder processes in respective responder nodes. Respective payloads of the MPI messages are combined in a network interface device of the initiator node to form an aggregated MPI message. The aggregated MPI message is transmitted through the fabric to network interface devices of responder nodes, disaggregating the aggregated MPI message into individual messages, and distributing the individual messages to the designated responder node processes.

Claims

1. A method of communication, comprising the steps of: executing a group of processes in a fabric of nodes comprising Network Interface Controllers (NICs), the NICs having respective NIC Communicator Controllers (NCCs); transmitting messages from all the processes to all the processes of the group of processes via the NICs by performing a Message Passing Interface (MPI) collective operation, wherein the nodes of the fabric function concurrently as initiator nodes executing respective initiator processes and as responder nodes executing respective responder processes, and wherein the collective operation comprises transmitting MPI messages through the fabric from all the initiator processes to all of the responder processes; with the NCCs of the initiator nodes combining respective payloads of the MPI messages to form an aggregated MPI message; transmitting the aggregated MPI message through the fabric to the responder nodes; in respective NCCs of the responder nodes disaggregating the aggregated MPI message into individual messages; and distributing the individual messages to the responder processes.

2. The method according to claim 1, wherein the aggregated MPI message has exactly one transport header that comprises a destination address of the aggregated MPI message.

3. The method according to claim 1, wherein the MPI messages comprise respective MPI headers comprising designations of the responder processes, wherein the responder processes are referenced in an MPI communicator object.

4. The method according to claim 3, wherein the MPI collective operation comprises forwarding by a communication library the MPI communicator object and the payloads to the NICs of the initiator nodes.

5. The method according to claim 1, further comprising: maintaining a communicator context in the NCCs of the initiator nodes, wherein transmitting the aggregated MPI message comprises directing the aggregated MPI message to local identifiers (LIDs) in the responder nodes according to the communicator context.

6. The method according to claim 1, comprising forming the aggregated MPI message by assembling pointers to message data, and including respective local identifier addresses for the message data in the aggregated MPI message.

7. An apparatus of communication, comprising: a fabric of nodes configured for executing a group of processes; respective network interface controllers (NICs) in the nodes; and respective NIC Communicator Controllers (NCCs) in the NICs, wherein the nodes are operative for transmitting messages from all the processes to all the processes of the group of processes via the NICs by performing a Message Passing Interface (MPI) collective operation, wherein the nodes of the fabric function concurrently as initiator nodes executing respective initiator processes and as responder nodes executing respective responder processes, and wherein the collective operation comprises transmitting MPI messages through the fabric from all the initiator processes to all of the responder processes; and wherein the NCCs of the initiator nodes are configured for combining respective payloads of the MPI messages to form an aggregated MPI message, and the NCCs of the responder nodes are configured for disaggregating the aggregated MPI message into individual messages, and wherein the responder nodes are operative for distributing the individual messages to the responder processes.

8. The apparatus according to claim 7, wherein the aggregated MPI message has exactly one transport header that comprises a destination address of the aggregated MPI message.

9. The apparatus according to claim 7, wherein the MPI messages comprise respective MPI headers containing designations of the responder processes, wherein the responder processes are referenced in an MPI communicator object.

10. The apparatus according to claim 9, wherein the MPI collective operation comprises forwarding by a communication library the MPI communicator object and the payloads to the NCCs of the initiator nodes.

11. The apparatus according to claim 7, wherein the first communicator controller circuitry is operative for forming the aggregated MPI message by assembling pointers to message data, and including respective local identifier addresses for the message data in the aggregated MPI message.

12. The apparatus according to claim 7, wherein the NCCs are operative for maintaining a communicator context and transmitting the aggregated MPI message by directing the aggregated MPI message to local identifiers (LIDs) in the responder nodes according to the communicator context.

Description

BRIEF DESCRIPTION OF THE SEVERAL VIEWS OF THE DRAWINGS

(1) For a better understanding of the present invention, reference is made to the detailed description of the invention, by way of example, which is to be read in conjunction with the following drawings, wherein like elements are given like reference numerals, and wherein:

(2) FIG. 1 is a schematic diagram of an exemplary computer system in which the principles of the invention are applied;

(3) FIG. 2 is a block diagram illustrating message aggregation and disaggregation by NICs in a fabric in accordance with an embodiment of the invention;

(4) FIG. 3 is a flow chart of a method of transmitting collectives through a fabric in an MPI job in accordance with an embodiment of the invention;

(5) FIG. 4 is a block diagram illustrating an aggregated message in accordance with an embodiment of the invention;

(6) FIG. 5 is a block diagram of a node in a fabric, which is configured for assembling and processing aggregated messages in accordance with an embodiment of the invention;

(7) FIG. 6 is a table illustrating an exemplary communicator context, in accordance with an embodiment of the invention;

(8) FIG. 7 is a detailed block diagram illustrating a process of alltoall aggregation and disaggregation in accordance with an embodiment of the invention; and

(9) FIG. 8 is a flow chart of the process of alltoall aggregation and disaggregation shown in FIG. 7 in accordance with an embodiment of the invention.

DETAILED DESCRIPTION OF THE INVENTION

(10) In the following description, numerous specific details are set forth in order to provide a thorough understanding of the various principles of the present invention. It will be apparent to one skilled in the art, however, that not all these details are necessarily always needed for practicing the present invention. In this instance, well-known circuits, control logic, and the details of computer program instructions for conventional algorithms and processes have not been shown in detail in order not to obscure the general concepts unnecessarily.

(11) Documents incorporated by reference herein are to be considered an integral part of the application except that, to the extent that any terms are defined in these incorporated documents in a manner that conflicts with definitions made explicitly or implicitly in the present specification, only the definitions in the present specification should be considered.

(12) Definitions.

(13) A switch fabric or fabric refers to a network topology in which network nodes interconnect via one or more network switches (such as crossbar switches), typically through many ports. The interconnections are configurable such that data is transmitted from one node to another node via designated ports. A common application for a switch fabric is a high performance backplane.

(14) System Architecture.

(15) Reference is now made to FIG. 1, which schematically illustrates an exemplary computer system 10 in which the principles of the invention are applied. The system 10 is configured for use in an InfiniBand fabric, but may be adapted for other networks by those skilled in the art. System 10 comprises nodes 10, 12, 14, 18, which are interconnected by a packet network 19, such as an InfiniBand switch fabric. In the pictured embodiment, node 12 is an initiator node for a collective operation, while nodes 14, 16, 18 are responder nodes, but typically any given node may be both an initiator and a responder concurrently.

(16) Reference is now made to FIG. 2, which is a block diagram illustrating message aggregation and disaggregation by NICs in a fabric 26 in accordance with an embodiment of the invention. The fabric 26 comprises a collective-initiating NIC 28, together with NICs 30, 32 and switches 34. Processes 36 (P_1 through P_P) execute in the host (not shown) of the NIC 28. NIC 28 is informed and keeps track of different communicators created in an MPI job. Other hosts similarly execute processes, including processes 38 in the host of responding NIC 32.

(17) Reference is now made to FIG. 3, which should be read in conjunction with FIG. 2. FIG. 3 is a flow chart of a method of transmitting collectives through a fabric in an MPI job in accordance with an embodiment of the invention. The process steps are shown in a particular linear sequence for clarity of presentation. However, it will be evident that many of them can be performed in parallel, asynchronously, or in different orders. Those skilled in the art will also appreciate that a process could alternatively be represented as a number of interrelated states or events, e.g., in a state diagram. Moreover, not all illustrated process steps may be required to implement the method.

(18) At initial step 40 an MPI alltoall or alltoallv collective operation is initiated by the host (not shown) of NIC 28. Next, at step 42 processes 36 (P_1 through P_P) commit their entire payloads to NIC 28. The payloads in this context are composed of all of the messages (including MPI headers) originated by the processes 36 to other processes in the communicator. These messages are referred to herein as MPI messages.

(19) After all local processes in the communicator have committed their alltoall payloads, at step 44 NIC 28 assembles a single message to each of the nodes in the communicator, referred to herein as an aggregated message. Reference is now made to FIG. 4, which is a block diagram illustrating an aggregated message 46 in accordance with an embodiment of the invention. The aggregated message 46 comprises a transport header 48, which specifies a destination address using a local identifier (LID). The aggregated message 46 also contains any number of MPI messages 50, each having an MPI header 52 with fields 54 that specify the processes of the relevant communicator that are relevant to the MPI messages 50. One way to create the aggregated message 46 is to traverse the alltoall payload and to aggregate all MPI messages that have the same destination address (LID).

(20) Reverting to FIG. 2 and FIG. 3, at step 56 the NIC 28 transmits the aggregated message 46 into the fabric, through the switches 34 toward its destination, designated by the LID as NIC 32 When the NIC 32 receives the aggregated message 46, at step 58 it disaggregates the aggregated message into individual MPI messages 50, and at final step 60 distributes the MPI messages 50 to their corresponding processes 38. Each of the processes 38 receives that component of the aggregated message 46 that pertains to it.

(21) Reference is now made to FIG. 5, which is a block diagram of a node 62 in a fabric, configured for assembling and processing aggregated messages in accordance with an embodiment of the invention. Elements above a broken line 64 are typically located in a host computer, while elements below the line 64 may be implemented in a network element, such as a network interface card. Although the node 62 is shown as comprising a number of separate functional blocks, these blocks are not necessarily separate physical entities, but rather represent different computing tasks or data objects stored in a memory that is accessible to a processor. These tasks may be carried out in software running on a single processor, or on multiple processors. The software may be embodied on any of a variety of known non-transitory media for use with a computer system, such as a diskette, or hard drive, or CD-ROM. The code may be distributed on such media, or may be distributed to the node 62 from the memory or storage of another computer system (not shown) over a network. Alternatively or additionally, the node 62 may comprise a digital signal processor, field programmable gate array or hard-wired logic. The node 62 is described with respect to an InfiniBand implementation, but can be applied to other network communications standards, mutatis mutandis.

(22) Any number of MPI processes 66 execute in the node 62. In this example all the MPI processes 66 are members of the same communicator. Instances of a communication software library 68 translate MPI commands of the MPI processes 66 into corresponding driver commands for a NIC driver 70. In an InfiniBand implementation, the MPI processes 66 translate the MPI commands into InfiniBand verb functions. The NIC driver 70 itself is a software library, which translates the driver commands issued by the library 68 into hardware commands that are acceptable to a network interface card 72. In an InfiniBand implementation the commands may be work queue elements (WQEs). Data aggregation and disaggregation (steps 44, 58; FIG. 3) are handled by a NIC communicator controller 74 (NCC) in a hardware communicator context when the node 62 acts as an initiator or responder as the case may be. MPI messages are received from the NIC communicator controller 74 and transmitted into the fabric by a packet sender 76. MPI messages are received from the fabric and delivered to the NIC communicator controller 74 by a packet receiver 78.

(23) Reference is now made to FIG. 6, which is a table 80 illustrating an exemplary communicator context in accordance with an embodiment of the invention. The table 80 assumes an InfiniBand implementation, but as noted above, it can be modified to accommodate other protocols. The table 80 facilitates handling data aggregation and disaggregation by the NIC communicator controller 74 (FIG. 5), and has two columns: an intra section 82 and an inter section 84. Entries in the intra section 82 are local MPI processes, including the queue pair (QP) that is assigned to each MPI process. Each MPI process commits its alltoall payload by writing a WQE to its local queue pair. Entries in the inter section 84 are destination addresses (DLID) of nodes having processes that are pertain to the communicator. Each DLID entry holds the number of MPI processes for that LID. The inter section 84 is used to send the alltoall payload from the local MPI processes to the different DLIDs.

(24) Reference is now made to FIG. 7, which is a detailed block diagram illustrating a process of alltoall aggregation and disaggregation in accordance with an embodiment of the invention. Three fabric nodes 86, 88, 90 are connected to switch 92. Processes 94, 96 execute in a host (not shown) of NIC 104. Process 98 (P_3) executes in a host (not shown) of NIC 106, and processes 100, 102 (P_4, P_5) execute in a host (not shown) of NIC 108. All the processes 94, 96, 98, 100, 102 share the same communicator. LIDS (labeled LID 5, LID 7, LID 12) are found in NICs 104, 106, 108. NICs 104, 106, 108 have NCCs 110, 112, 114, and communicator contexts 116, 118, 120, respectively. NIC 104 is associated with queue pair 122, 124 (Qp 1; Qp 2). A work queue element 126 associated with queue pair 122 is established and includes a data pointer 128 to alltoall data (represented by a block 130). NIC 104 may also generate another queue pair 132 (Qp100), whose function is explained below. NIC 108 is associated with queue pairs 134(Qp 4), 136 (Qp 5).

(25) Reference is now made to FIG. 8, which is a flow chart of the process of alltoall aggregation and disaggregation shown in FIG. 7 in accordance with an embodiment of the invention. At initial step 138 an alltoall operation is begun.

(26) Next, at step 140 the communicator contexts 116, 118, 120 are initialized on their respective NICs 104, 106, 108, with the corresponding fields that describe the communicator and are associated with respective queue pairs. For example, on LID 7 in NIC 104, the local MPI process queue pairs are queue pairs 122, 124 and the remote LIDs are LID 5 and LID 12 in NIC 106 108, respectively. LID 12 in NIC 108 contains two MPI processes 100, 102.

(27) Next, at step 142 the MPI alltoall function is invoked by all of the local MPI processes of the node 86.

(28) Next, at step 144 the communicator and the alltoall payload are forwarded to the NIC 104 by the communication library. In an Infiniband implementation, step 144 is comprises posting work queue element 126 to queue pair 122, 124, which, as noted above, includes data pointer 128 to the payload data in block 130. In the example of FIG. 7, process 94 (P_1) has an alltoall payload for processes 98, 100, 102 (P_3-P_5), which is accessed in the NIC 104 using data pointer 128.

(29) Next, at delay step 146 the NCCs 110, 112, 114 in the NICs 104, 106, 108 wait for all of the MPI processes to commit their alltoall payloads. For example, NIC 104 waits for queue pair 122, 124 to post the work queue element 126.

(30) After all local processes have committed their data, at step 148 the NCC 110 assembles the data pointers and creates a single aggregated message, which is directed to the LIDS in the remote NICs 106, 108 according to the communicator context. The NCC 110 is aware of the organization of the alltoall data, and thus which data belong to which LID. In an InfiniBand implementation, the NCC 110 may use a different queue pair from the queue pairs of the local processes to transmit the data. The NCC 110 may also add an extra header to the aggregated message in order to identify the communicator on which the alltoall operation is performed.

(31) In the above example, queue pair 132 is used to send the data, and the message transfer comprises two messages: one message to LID 5 in NIC 106 containing alltoall data for remote process 98 and one message to LID 12 in NIC 108 containing alltoall data for the remote process 100, 102.

(32) The aggregated message is transmitted at step 150 When the aggregated message arrives at its destinations, the communicator contexts 118, 120 are fetched again at step 152 by the receiving NCCs 112, 114, respectively. The NCCs 112, 114 are aware of the order of the alltoall payload of the aggregated message.

(33) Then, at final step 154 the NCCs 112, 114 disaggregate the aggregated message and scatter the data to the MPI processes according to the communicator contexts 118, 120, respectively. In above example, the NCC 114 in NIC 108 breaks the message into two parts, and scatters the first half to queue pair 134 (Qp 4) and the second half to queue pair 136 (Qp 5).

(34) It will be appreciated by persons skilled in the art that the present invention is not limited to what has been particularly shown and described hereinabove. Rather, the scope of the present invention includes both combinations and sub-combinations of the various features described hereinabove, as well as variations and modifications thereof that are not in the prior art, which would occur to persons skilled in the art upon reading the foregoing description.