RUNTIME OPTIMIZATION OF DISTRIBUTED EXECUTION GRAPH

• US 20080098375A1
• Filed: 09/29/2006
• Published: 04/24/2008
• Est. Priority Date: 09/29/2006
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Corporation
• Inventors: Isard, Michael A.
• Granted Patent: US 7,844,959 B2
• Time in Patent Office: Days
• US Class Current: 717/149
• CPC Class Codes: G06F 11/1438 Restarting or rejuvenating
• CPC Class Codes: G06F 9/5066 Algorithms for mapping a plurality of inter-dependent sub-tasks onto a plurality of physical CPUs mappping at compile time, see G06F8/451

Abstract

A general purpose high-performance distributed execution engine for coarse-grained data-parallel applications is proposed that allows developers to easily create large-scale distributed applications without requiring them to master concurrency techniques beyond being able to draw a graph of the data-dependencies of their algorithms. Based on the graph, a job manager intelligently distributes the work load so that the resources of the execution engine are used efficiently. During runtime, the job manager (or other entity) can automatically modify the graph to improve efficiency. The modifications are based on runtime information, topology of the distributed execution engine, and/or the distributed application represented by the graph.

Claims

1. A method for performing distributed parallel processing, comprising:

assigning a first set of code corresponding to vertices of a graph to various nodes of a distributed parallel processing engine based on said graph, said graph defines a parallel processing job;

automatically modifying said graph based on executing said first set of code; and

assigning a second set of code corresponding to vertices of said graph to various nodes of said distributed parallel processing engine based on said modified graph, said first set of code and said second set of code are part of said parallel processing job.

2. A method according to claim 1, wherein said automatically modifying said graph comprises:

determining where in a network said first set of code executed;

grouping together at least a first subset of said first set of code that executed in a common section of said network; and

adding additional code to process data from said first subset, said additional code corresponds to a new vertex on said graph, said additional code is assigned to run in said common section.

3. A method according to claim 2, wherein said automatically modifying said graph further comprises:

providing a data channel to said additional code from existing code corresponding to an existing original node in said graph;

causing said additional code to be executed in said common section of said network.

4. A method according to claim 2, wherein:

said additional code is a copy of code for an existing vertex.

5. A method according to claim 2, wherein:

said additional code includes code for a new vertex not existing in said graph prior to said modifying.

7. A method according to claim 1, wherein:

said automatically modifying said graph based on executing said first set of code takes into account volume of data processed by said first set of code.

8. A method according to claim 1, further comprising:

receiving one or more indications of one or more portions of said graph that are eligible to be modified, said automatically modifying said graph is restricted to said one or more portions based on said one or more indications.

9. A method according to claim 1, wherein:

said graph is a directed acyclic graph.

6. A method according to claim 6, wherein:

said additional code, said first set of code and said graph are provided by a developer.

10. One or more processor readable storage devices having processor readable code stored thereon, said processor readable code programs one or more processors to perform a method comprising:

accessing a representation of a directed acyclic graph defining a parallel processing job, said graph having vertices and edges connecting to said vertices, said vertices represents sets of program code of said parallel processing job to be run on a distributed execution engine;

automatically modifying said graph during runtime of said parallel processing job on said distributed execution engine based on runtime information of said parallel processing job on said distributed execution engine; and

assigning code corresponding to vertices of said graph to various nodes of said distributed parallel processing engine based on said graph.

11. One or more processor readable storage devices according to claim 10, wherein:

said graph includes a first set of vertices at a first level of hierarchy in communication with a second set of vertices at a second level of hierarchy;

said automatically modifying said graph includes grouping a first subset of said first set of vertices into a first group and grouping a second subset of said first set of vertices into a second group based on location and availability of nodes on said distributed execution engine; and

said automatically modifying said graph further includes adding additional vertices of said second set of vertices to said first group.

12. One or more processor readable storage devices according to claim 10, wherein said automatically modifying comprises:

determining where a first set of vertices were executed in said distributed execution engine;

grouping together vertices of said first set based on where they ran in said distributed execution engine;

adding new vertices to each group; and

adding new edges for said new vertices.

13. One or more processor readable storage devices according to claim 12, wherein said automatically modifying further comprises:

recursively grouping vertices and adding new vertices,

14. One or more processor readable storage devices according to claim 10, wherein said automatically modifying comprises:

determining that a first set of vertices were executed in a portion of said distributed execution engine associated with a first switch;

grouping together a first set of vertices;

adding a new vertex to said graph;

adding new edges connecting said new vertex in said graph; and

causing code for said new vertex to run in said portion of said distributed execution engine associated with said first switch.

15. One or more processor readable storage devices according to claim 14, wherein:

said new vertex is a copy of an existing vertex.

16. One or more processor readable storage devices according to claim 11, wherein said method further comprises:

receiving an indication of a portion of said graph that is eligible to be modified, said automatically modifying said graph is restricted to said portion based on said indication.

17. A distributed parallel processing system, comprising:

a network;

a plurality of computing machines connected to said network; and

a job manager connected to said network and in communication with said computing machines, said job manager manages execution of a job defined by a user customizable graph, said user customizable graph includes a set of vertices corresponding to a set of program units and edges corresponding to data channels, said job manager assigns said program units for execution on said computing machines based on said graph;

said job manager automatically modifies said graph during runtime of said job based on topology of said network, availability of said computing machines and said vertices.

18. A distributed parallel processing system according to claim 17, wherein:

said job manager automatically modifies said graph by determining where a first set of vertices were executed in said network, grouping together vertices of said first set based on where they ran in said network;

adding new vertices to each group, and adding new edges for said new vertices.

19. A distributed parallel processing system according to claim 17, wherein:

said job manager only modifies one or more portions of said graph that are previously indicated by a user to be eligible for modification.

20. A distributed parallel processing system according to claim 17, further comprising:

a name server connected to said network and in communication with said job manager, said name server informs said job manager of identifications of said computing machines, said name server provides information to said job manager about topology of said network, said network is a tree-structured network.