Methods and apparatus to perform process placement for distributed applications

Methods and apparatus to perform process placement for distributed applications are disclosed. An example method comprises determining a mapping between a communication graph representative of communications of a distributed application and a topology graph representative of communication costs associated with a computing network, and executing the distributed application with the processes of the distributed application assigned to the processing entities of the computing network based upon the mapping.

FIELD OF THE DISCLOSURE

This disclosure relates generally to distributed applications and, more particularly, to methods and apparatus to perform process placement for distributed applications.

BACKGROUND

A message-passing parallel application (i.e., a distributed application) is cooperatively implemented via generally contemporaneous execution of two or more machine accessible instructions (e.g., processes) by one or more processors and/or cores. A distributed application often has a non-uniform number of messages and/or data to be communicated between the two or more of processes that collectively implement the distributed application.

Symmetric Multi-Processor (SMP) clusters, multi-clusters and/or computing networks are commonly used to execute and/or implement distributed applications. Such computing networks often have non-uniform communication costs associated with the transmission of messages and/or data between the processors, cores and/or computing nodes that form the computing network. For instance, an example computing node contains multiple processors and/or cores and has high bandwidth and/or low latency (i.e., low communication cost) communication paths that connect the processors and/or cores. However, communication paths between processors and/or cores of this example computing node and another processor and/or core associated with any other computing node may have substantially lower bandwidth and/or substantially higher latency (i.e., a higher communication cost). For example, messages and/or data passed between two computing nodes may traverse through multiple Ethernet switches and/or communication links and, thus, exhibit relatively higher latency and/or lower bandwidth.

Given the non-uniformity of communication requirements for a distributed application and the non-uniformity of communication costs for a computing network, the assignment of processes of a distributed application to processors, cores and/or computing nodes of a computing network has a direct and/or potentially significant impact on the performance (e.g., execution speed) of the distributed application.

DETAILED DESCRIPTION

FIG. 1is a schematic illustration of an example system to perform process mapping for one or more distributed applications. In the example system ofFIG. 1, an example distributed application is cooperatively implemented via generally contemporaneous execution of machine accessible instructions by two or more processors and/or cores of a computing network105. For example, a first process (e.g., a software application or portion of a software application) executed by a first processor and/or core, a second process executed by a second processor and/or core, a third process executed by a third processor and/or core, etc. cooperatively realize a distributed application using any variety of distributed computing algorithms, techniques and/or methods. The various processes of a distributed application may implement different, similar and/or identical machine accessible instructions. Moreover, more than one process may be implemented by any particular processor and/or core. Further, any number of processors and/or cores (e.g., 2, 3, 4, etc.) may be used to execute a distributed application.

In the example system ofFIG. 1, one or more processors and/or cores are implemented within a computing node (e.g., a dual-processor and/or dual-core computer, server and/or workstation) with a plurality of computing nodes forming the example computing network105. For simplicity, the term processing entity will be used herein to refer to processors, cores and/or computing nodes. The processes of a distributed application may be developed using any variety of programming tool(s) and/or language(s) and may be used to implement any variety of distributed application(s). Further, example processing entities of the example computing network105ofFIG. 1may execute any variety of operating system(s). It will be readily appreciated by persons of ordinary skill in the art that the methods and apparatus to perform process mapping disclosed herein may be applied to any type, topology and/or size of computing networks105and/or to any variety of distributed applications.

To characterize the communication requirements for an example distributed application, the example system ofFIG. 1includes a communication profiler110. Example communication requirements include a number of messages, a number of bytes, etc. sent between any two of the processes implementing the example distributed application for, for example, a representative time period, function(s), etc. In the illustrated example ofFIG. 1, the example communication profiler110profiles the communication requirements of the example distributed application while the distributed application is executing on the example computing network105. Using any variety of method(s), technique(s), application programming interface(s) and/or user interfaces(s), the communication profiler110analyzes trace information collected by any variety of tracing tool115such as, for example, the Intel® Trace Analyzer and Collector or the Intel® message passing interface (MPI) library. Alternatively, the example communication profiler110may characterize the distributed application by analyzing the source code of the distributed application and/or by relying on information and/or parameters provided by, for example, a programmer of the distributed application.

It will be readily apparent to persons of ordinary skill in the art that the communication requirements for a distributed application may vary. That is, the communication requirements for a first portion of a distributed application may be different than those for a second portion. As such, the example communication profiler110ofFIG. 1may be used to profile all or any portion of a distributed application. For example, the communication profiler110may be used to profile a portion representing the substantially largest communication needs and/or computational processing. The communication profiler110may also be used to profile an entire distributed application and, thus, the communication requirements represent a sort of overall average of the communication requirements. Moreover, if a distributed application is modified (e.g., changed number of processes, application is scaled, re-distribution of workload amongst the processes, etc.), its communication requirements may change and, thus, it may be required, desired and/or beneficial for the communication profiler110to re-determine the communication requirements for the modified distributed application.

The example communication profiler110ofFIG. 1compiles the communication requirements into a communication graph120having a plurality of graph edges that represents the communication requirements between each pair of the processes that implement the example distributed application. In the example ofFIG. 1, the example communication graph120is stored as, for example, a data structure (e.g., a matrix, an array, variable(s), register(s), a data table, etc.) in, for example, a memory and/or a machine accessible file122that is accessible to a graph mapper125. An example data structure to store a communication graph120is discussed below in connection withFIG. 2A.

To characterize the communication costs associated with the example computing network105, the example system ofFIG. 1includes a network profiler130. Example communication costs include a maximum bandwidth, a latency (e.g., microsecond per kilo byte (Kbyte)), an overhead, etc. between each pair of the processing entities (e.g., processors, cores, computing nodes, etc.) that implement the example computing network105. The example network profiler130ofFIG. 1profiles the communication costs of the example computing network105using any variety of topology discovery mechanism(s), method(s) and/or technique(s) such as, for example, any variety and/or combination of a message-passing parallel ping-pong tool, a trace collector and/or an MPI library. For example, a trace collector could be used to characterize a message-passing parallel ping-pong tool, thus, discovering the topology of a computing network. For example, outputs of the message-passing parallel ping-pong tool could be used to directly characterize the communication costs associated with the topology. Additionally or alternatively, the example network profiler130ofFIG. 1could characterize the communication costs based upon a priori information regarding the communication device(s), communication paths and/or communication links used to connect the processing entities of the example computing network105. Example a priori information includes a bus transfer speed, the delay and/or latency through an Ethernet and/or ATM switch, etc.

It will be readily apparent to persons of ordinary skill in the art that if the size, topology, etc. of the example computing network105is altered, changed and/or, otherwise modified, its communication costs may change and, thus, it may be desired and/or beneficial for the network profiler130to re-determine the communication costs for the modified computing network105. Moreover, the communication costs may change over time depending on, for example, whether and/or how other distributed application(s), processes, jobs, etc. are running and/or scheduled on the example computing network105.

The example network profiler130ofFIG. 1compiles the communication costs into a topology graph135having a plurality of graph edges that represents the communication requirements between each pair of the processing entities that implement the example computing network105. In the example ofFIG. 1, the example topology graph135is stored as, for example, a data structure (e.g., a matrix, an array, variable(s), register(s), a data table, etc.) in, for example, a memory and/or a machine accessible file137that is accessible to the graph mapper125. An example data structure to store a communication graph135is discussed below in connection withFIG. 2B.

To determine a mapping between processes of an example distributed application and processing entities of the example computing network105, the example system ofFIG. 1includes the graph mapper125. The example graph mapper125ofFIG. 1determines a mapping of the vertices of the communication graph120for the example distributed application to the vertices of the topology graph135for the computing network105that reduces the total and/or overall communication cost for the example distributed application. In the illustrated example ofFIG. 1, for a particular mapping of processes (i.e., nodes of the communication graph120) to processing entities (i.e., nodes of the topology graph135), the total and/or overall communication cost of a distributed application is computed as the sum of the costs associated with each of the edges resulting from a particular mapping. The example graph mapper125uses, for example, a linear matrix M that is indexed with the numbers of processes to represent the mapping between processes and processing entities. An example matrix M=[1, 3, 2, 4] corresponds to the example mapping illustrated and discussed below in connection withFIG. 2C. For purposes of explanation, a linear matrix M will be used herein, however, persons of ordinary skill in the art will readily recognize that any other variety of data structure, array, matrix, variable(s), register(s) and/or table could be used to represent a mapping between processes and processing entities.

In the example system ofFIG. 1, the cost of a resulting mapped edge is computed using any variety of method(s) and/or technique(s) such as, for example, multiplying the associated communication requirements and communication costs. The example graph mapper125ofFIG. 1locates a mapping that reduces the sum of these resulting map edge costs. In particular, the example graph mapper125locates a mapping that representing a minima of the following mathematical expression:
Σf(Wij,dkl),   EQN. 1

where wijis the communication graph edge value between processes i and j, dk,lis the topology graph edge value between processing entities k and l, where k=M[i] and l=M[j], and f( ), for example, is a function that multiples the two values wijand dk,l.

Starting with an initial random mapping M, the example graph mapper125sequentially considers alternative mappings. In particular, the example graph mapper125ofFIG. 1considers alternative mappings that result from a switch of the mapping of two processes. For example, if a first mapping maps processes i and j to processing entities M[i] and M[j], respectively, an example alternative mapping maps processes i and j to processing entities M[j] and M[i], respectively. The improvement and/or decrements (i.e., gain) resulting from such a mapping switch can be computed as a difference of the value of the mathematical expression of EQN. 1 before and after the considered pair switch. In particular, components gain(i,j) of a gain matrix that represents the swapping of all pairs of processes i and j can be computed using the following mathematical expression:

Starting with an initial random mapping M, the example graph mapper125uses the following process to locate the lowest overall cost mapping of processes to processing entities. The example graph mapper125first computes the gain matrix using EQN. 2 and then selects a process pair swap that results in the largest gain (i.e., the maximum gain matrix entry) and has processes that have not yet been swapped. The example graph mapper125saves the gain matrix entry (i.e., the gain that would result from a swap of the selected process pair) and then recalculates the entire gain matrix to model the mapping if the process swap was made. The example graph mapper125continues selecting process pairs to swap and re-computing the gain matrix until all of the pairs of the processes of the distributed application (i.e., vertices of the communication graph120) have been swapped. The example graph mapper125then determines which of the pair swaps resulted in the largest saved gain. The pair swap providing the largest saved gain is retained and all others swaps are discarded. The example graph mapper125repeats the process described above until no additional swaps can be identified that result in an improvement to the overall communication cost for the distributed application (i.e., a local minima has been identified). As discussed below, to reduce the likelihood of finding a local minima, as opposed to an overall minima, the process may be repeated starting from one or more additional random mappings and then selecting the result that provides the lowest overall communication cost.

In the illustrated example ofFIG. 1, the number of processes of the distributed application and the number of processing entities of the example computing network105are equal. If the number of processes is not equal to the number of processing entities then dummy vertices can be inserted into the smaller of the communication graph or the topology graph to equalize the sizes of the matrices.

The example graph mapper125ofFIG. 1stores the resulting graph mapping140into any variety of data structure (e.g., a matrix, an array, variable(s), register(s), a data table, etc.) in, for example, a memory and/or a machine accessible file142that is accessible to any variety of software entity and/or tool associated with and/or a part of the example computing network105that is responsible for setup and initialization of a distributed application. In the example ofFIG. 1, the mapping data140is simply a list associating particular processes of the distributed application with particular processing entities of the example computing network105. In the example ofFIG. 1, the Intel® Cluster Toolkit is used to read the mapping data140and to setup and/or initialize the distributed application based upon the mapping of processes to processing entities determined by the example graph mapper125.

It will be readily apparent to persons of ordinary skill in the art that the memories and/or machine accessible files122,137and/or142may be implemented using any number of memories and/or machine accessible files. For example, a single memory may be used to store the communication graph120, the topology graph135and the mapping data140.

To measure the performance of a distributed application, the example system ofFIG. 1includes any variety of performance profiler145. Using any variety of technique(s) and/or method(s), the example performance profiler145ofFIG. 1determines the execution speed (e.g., in seconds) and/or bandwidth (e.g., Mega-flops per second) of the distributed application. For example, the performance profiler145may be used to measure the performance improvement of a distributed application resulting from a process to processing entity mapping.

Although an example system to map processes of a distributed application to processing entities of a computing network105and to execute the distributed application based on the mapping has been illustrated inFIG. 1, distributed application systems may be implemented using any of a variety of alternative and/or additional devices, entities, modules, etc. Further, the devices, entities, modules, elements, etc. illustrated inFIG. 1may be combined, re-arranged, and/or implemented in any of a variety of ways. For example, the communication profiler110and tracing tool115may be implemented using a single computing device and/or platform. Further still, any or all of the example tracing tool115, the example communication profiler110, the example graph mapper125, the example network profiler130and/or the example performance profiler145may be implemented by hardware, software, firmware and/or any combination of hardware, software and/or firmware.

FIG. 2Aillustrates an example communication graph120for an example distributed application. The example distributed application ofFIG. 2Aincludes of four (4) processes P1, P2, P3and P4that form the vertices of the example communication graph120ofFIG. 2A. The numbers associated with the graph edges illustrated inFIG. 2Arepresent the communication requirements between the two processes at either end of the graph edge. An example graph edge205between processes P1and P2requires 20 units of communication (e.g., messages and/or bytes). Alternatively or additionally, the communication requirements associated with each communication graph edge may represent, for example, a transmission and/or delay time that does not undesirably slow execution of the distributed application. In the example system ofFIG. 1, communication requirements are inversely proportional delay times. For example, a short delay time corresponds to a large communication requirement. In the illustrated example ofFIG. 2A, there are no communication requirements between processes P2and P3or between processes P1and P4.

FIG. 2Billustrates an example topology graph135for an example computing network105. The example computing network105ofFIG. 2Bincludes of four (4) processing entities (e.g., cores) N1, N2, N3and N4that form the vertices of the example topology graph135ofFIG. 2B. The numbers associated with the graph edges illustrated inFIG. 2Brepresent the communication costs between the two cores at either end of the graph edge. An example graph edge210between cores N1and N3represents a latency of 20 units of time (e.g., seconds). Alternatively or additionally, the communication costs associated with each topology graph edge may represent, for example, a bandwidth (e.g., bytes per second) associated with the corresponding processing entity pair, a latency (i.e., delay in communication), etc. Since, in the example ofFIG. 2A, cores N1and N2are implemented within a single computing node212(e.g., within a single semiconductor package), the latency215between them is lower than, for example, the latency210between the cores N1and N3that are implemented in separate semiconductor packages and/or computing nodes.

FIG. 2Cillustrates an example mapping of the example communication graph120ofFIG. 2Ato the topology graph135ofFIG. 2Bthat reduces the overall distributed application communication cost computed using, for example, EQN. 1. The example graph mapper125ofFIG. 1determines the example mapping ofFIG. 2Cby executing, for example, the machine accessible instructions discussed below in connection withFIG. 4. As illustrated inFIG. 2C, process P1is mapped to core N1, process P2is mapped to core N3, process P3is mapped to core N2and process P4is mapped to core N4. The numbers associated with the graph edges ofFIG. 2Crepresent the resulting communication cost between the two mapped processes at either end of the graph edge. An example graph edge220between process P1mapped to core N1and process P2mapped to core N3represents a communication cost equal to the product of the communication requirement205ofFIG. 2Aand the communication cost210ofFIG. 2B.

While the methods disclosed herein do not directly identify deficiencies and/or beneficial changes to a computing network, the resulting communication costs (e.g., the edges ofFIG. 2C) associated with the mapping of a communication graph (e.g.,FIG. 2A) to a topology graph (e.g.,FIG. 2B) may be used by, for example, a programmer and/or analysis program and/or process to identify one or more ways that a computing network and/or distributed application could be alternated, changed, enhanced to improve the performance of the mapped distributed application. For example, the resulting communication costs could be used to determine the benefit of adding additional process(es), additional processing entity(ies), additional communication link(s), etc. Moreover, the methods disclosed herein could, additionally or alternatively, be used to evaluate and/or characterize possible performance and/or communication improvements resulting from a change in a distributed application and/or computing network.

FIG. 3Ais an example matrix (i.e., a data structure) that represents the example communication graph120ofFIG. 2A. The example matrix is a square matrix with each of the process nodes P1, P2, P3and P4ofFIG. 2Acorresponding to both a row and a column of the example data matrix. An example entry305in the 1strow (corresponding to process P1) and 2ndcolumn (corresponding to process P2) corresponds to the communication requirement205between processes P1and P2ofFIG. 2A. While the example data matrix ofFIG. 3Ais used to represent the example communication graph120ofFIG. 2A, persons of ordinary skill in the art will readily recognize that any other variety of data structure, array, matrix, variable(s), register(s) and/or table could be used to represent a communication graph.

FIG. 3Bis an example matrix (i.e., a data structure) that represents the example topology graph135ofFIG. 2B. The example matrix is a square matrix with each of the cores N1, N2, N3and N4ofFIG. 2Bcorresponding to both a row and a column of the example data matrix. An example entry310in the 2ndrow (corresponding to core N2) and 4thcolumn (corresponding to core N4) corresponds to the communication cost between cores N2and N4ofFIG. 2B. While the example data matrix ofFIG. 3Bis used to represent the example topology graph135ofFIG. 2B, persons of ordinary skill in the art will readily recognize that any other variety of data structure, array, matrix, variable(s), register(s) and/or table could be used to represent a topology graph.

FIG. 4is a flowchart representative of example machine accessible instructions that may be executed to implement the example graph mapper125ofFIG. 1. The example machine accessible instructions ofFIG. 4may be executed by a processor, a core, a controller and/or any other suitable processing device. For example, the example machine accessible instructions ofFIG. 4may be embodied in coded instructions stored on a tangible medium such as a flash memory, or random access memory (RAM) associated with a processor (e.g., the processor710shown in the example processor platform700and discussed below in conjunction withFIG. 7). Alternatively, some or all of the example flowchart ofFIG. 4may be implemented using an application specific integrated circuit (ASIC), a programmable logic device (PLD), a field programmable logic device (FPLD), discrete logic, hardware, firmware, etc. Also, some or all of the example flowchart ofFIG. 4may be implemented manually or as combination(s) of any of the foregoing techniques, for example, a combination of firmware, software and/or hardware. Further, although the example machine accessible instructions ofFIG. 4are described with reference to the flowchart ofFIG. 4, persons of ordinary skill in the art will readily appreciate that many other methods-of implementing the example graph mapper125ofFIG. 1may be employed. For example, the order of execution of the blocks may be changed, and/or some of the blocks described may be changed, eliminated, sub-divided, or combined. Additionally, persons of ordinary skill in the art will appreciate that the example machine accessible instructions ofFIG. 4may be carried out sequentially and/or carried out in parallel by, for example, separate processing threads, processors, devices, circuits, etc. Moreover, the machine accessible instructions ofFIG. 4may be carried out, for example, in parallel with any other variety of processes, while the distributed application is executing, etc.

The example machine accessible instructions ofFIG. 4begin with the graph mapper125reading and/or accessing the communication graph120for a particular distributed application (block405) and reading and/or accessing the topology graph135for a particular computing network105to which the distributed application is to be mapped (block410). To increase the likelihood of locating the best solution, as opposed to a local minimum, the graph mapper125creates an initial random mapping M of the processes to the processing entities (block415).

The graph mapper125then calculates the entries of a gain matrix for the initial mapping using, for example, the mathematical expression of EQN. 2 (block420). The graph mapper125then locates the matrix entry having the largest value and not corresponding to a process that has already been temporarily swapped (block425). The graph mapper125saves the identified matrix entry (i.e., the gain that would result if the processes were swapped) (block430) and temporarily swaps the corresponding entries in the mapping matrix M (block435). The graph mapper125then recalculates all of the entries of the gain matrix using, for example, the mathematical expression of EQN. 2 (block440). If not all processes have been temporarily swapped (block445), control returns to block425to locate the matrix entry having the largest value and not corresponding to a process that has been temporarily swapped.

When all processes have been temporarily swapped (block445), based on the matrix entries saved at block430(i.e., gains for each of the temporary process swaps), the graph mapper125determines which process mapping swap resulted in the largest gain (block450). If the gain due to the selected swap is positive (block455), the graph mapper125discards all of the temporary process swaps except for the swap having the largest saved gain (block460). That is the graph mapper125changes back the changes temporarily made to the mapping M while retaining the swap having the largest gain. Control then returns to block420to repeat the process. If the gain due to the selected swap is less than or equal to zero (block455), the graph mapper125discards all of the temporary process swaps since the prior mapping already represented a local minima. The example machine accessible instructions ofFIG. 4are then ended.

Alternatively, after block465the example graph mapper125could save the current mapping and control could then return to block415to locate another mapping starting from another initial random mapping. The better of the two mappings (i.e., the mapping providing the lowest overall communication cost) could then be selected. The graph mapper125could repeat this process to determine any number of candidate mappings using any number of initial mappings. For example, all possible mappings could be tested, in which case, the initial mapping need not be random.

FIG. 5Aillustrates an example two-tier computing network105including of eight (8) computing nodes505that are communicatively coupled via an Ethernet switch510. Each of the example computing nodes505ofFIG. 5Ainclude two processors and/or cores. In the example ofFIG. 5A, there will be a substantially higher communication costs for communications between computing nodes505than between processors and/or cores within a given computing node505.

FIG. 5Billustrates an example three-tier computing network105including of the eight (8) computing nodes505ofFIG. 5A. In contrast toFIG. 5A, the computing nodes505are communicatively coupled via two levels of Ethernet switches in the example ofFIG. 5B. A first set of four (4) of the computing nodes515are communicatively coupled to a first Ethernet switch520, while a second set of four (4) of the computing nodes525are communicatively coupled to a second Ethernet switch530. The Ethernet switches520and530are communicatively coupled via a third Ethernet switch535. In the illustrated example ofFIG. 5B, communication cost increase as messages and/or data pass between additional Ethernet switches. For example, the communication cost between two computing nodes attached to the same Ethernet switch (e.g., two nodes in the subset515) will be lower than the communication cost for data that has to pass through all three (3) Ethernet switches520,530and535(e.g., between a node of the subset515and a node of the subset525).

FIGS. 6A and 6Billustrate performance improvements resulting from the graph mapping methods and apparatus described above for a variety of industry-standard benchmark distributed applications605.FIG. 6Aillustrates the performance610resulting from a default mapping of processes to processors and/or cores for the example two-tier computing network105ofFIG. 5Afor each of the applications605. Also illustrated in FIG.6A is the performance615and speedup620that result when the processes of the distributed applications605are mapped to processors and/or cores of the two-tier network ofFIG. 5Avia the example process ofFIG. 4to reduce the overall communication costs of the distributed application. Likewise,FIG. 6Billustrates the performance625resulting from a default mapping of processes to processors and/or cores for the example three-tier computing network105ofFIG. 5B. Also illustrated inFIG. 6Bis the performance630and speedup635that result when the processes of the distributed applications605are mapped to processors and/or cores of the three-tier example network ofFIG. 5Bvia the example process ofFIG. 4to reduce the overall communication costs of the distributed application.

FIG. 7is a schematic diagram of an example processor platform700that may be used and/or programmed to implement the example communication profiler110, the example tracing tool115, the example graph mapper125, the example network profiler130and/or the example performance profiler ofFIG. 1. For example, the processor platform700can be implemented by one or more general purpose processors, cores, microcontrollers, etc.

The processor platform700of the example ofFIG. 7includes a general purpose programmable processor710. The processor710executes coded instructions727present in main memory of the processor710(e.g., within a RAM725). The processor710may be any type of processing unit, such as a processor from the Intel® families of processors. The processor710may execute, among other things, the example machine accessible instructions ofFIG. 4to implement the example graph mapper125ofFIG. 1.

The processor710is in communication with the main memory (including a read only memory (ROM)720and the RAM725) via a bus705. The RAM725may be implemented by dynamic random access memory (DRAM), Synchronous DRAM (SDRAM), and/or any other type of RAM device, and ROM may be implemented by flash memory and/or any other desired type of memory device. Access to the memory720and725is typically controlled by a memory controller (not shown) in a conventional manner. The RAM725may be used to store, for example, the example communication graph120and/or the example topology graph135.

The processor platform700also includes a conventional interface circuit730. The interface circuit730may be implemented by any type of well-known interface standard, such as an external memory interface, serial port, general purpose input/output, etc.

One or more input devices735and one or more output devices740are connected to the interface circuit730. For example, the input devices735may be used to provide and/or output the example mapping data140.