SAVING PROGRAM EXECUTION STATE

• US 20100153955A1
• Filed: 12/12/2008
• Published: 06/17/2010
• Est. Priority Date: 12/12/2008
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Nowland, Ian P., Cole, Richard J., Cabrera, Luis Felipe, Sirota, Peter, Khanna, Richendra
• Granted Patent: US 8,370,493 B2
• Time in Patent Office: Days
• US Class Current: 718/102
• CPC Class Codes: G06F 11/1451 by selection of backup contents
• CPC Class Codes: G06F 11/1469 Backup restoration techniques
• CPC Class Codes: G06F 9/4806 Task transfer initiation or dispatching
• CPC Class Codes: G06F 9/485 Task life-cycle, e.g. stopping, restarting, resuming execution G06F9/4881 takes precedence

Abstract

Techniques are described for managing distributed execution of programs. In at least some situations, the techniques include decomposing or otherwise separating the execution of a program into multiple distinct execution jobs that may each be executed on a distinct computing node, such as in a parallel manner with each execution job using a distinct subset of input data for the program. In addition, the techniques may include temporarily terminating and later resuming execution of at least some execution jobs, such as by persistently storing an intermediate state of the partial execution of an execution job, and later retrieving and using the stored intermediate state to resume execution of the execution job from the intermediate state. Furthermore, the techniques may be used in conjunction with a distributed program execution service that executes multiple programs on behalf of multiple customers or other users of the service.

Claims

1. A method for a computing system of a distributed program execution service to manage distributed execution of programs, the method comprising:

under control of the computing system of the distributed program execution service, the distributed program execution service providing a plurality of computing nodes that are configurable to execute programs of a plurality of users, receiving multiple requests to execute indicated programs using indicated input data, each of the requests being from one of the plurality of users and including indications of the program and the input data to be used for the request, and automatically responding to each request by;

automatically selecting multiple of the plurality of computing nodes for use in executing multiple execution jobs of the indicated program for the request in a distributed parallel manner, each of the multiple execution jobs having multiple operations to be performed using a subset of the indicated input data that is selected for the execution job;

for each of the multiple computing nodes, initiating execution on the computing node of one of the multiple execution jobs using the selected subset of input data for the one execution job, the initiating of the execution including configuring a portion of a distributed file system on the computing node for use in locally storing intermediate output data that is generated by completed performance of one or more of the multiple operations for the one execution job;

before the execution of at least some of the multiple execution jobs on at least some of the multiple computing nodes is completed,

automatically monitoring a state of partial execution of each of the at least some execution jobs, the monitoring of each of the at least some execution jobs including identifying intermediate output data that is locally stored on the distributed file system portion for the computing node on which the execution job executes, the intermediate output data being generated by one or more operations of the execution job whose performance is completed; and

determining to terminate execution at a first time of one or more of the at least some execution jobs, and in response to the determining and for each of the one or more execution jobs, automatically initiating remote persistent storage of the intermediate output data that is stored on the distributed file system portion for the computing node on which the execution job executes;

at a later second time after the first time, for each of the one or more execution jobs, initiating a resumed execution of the execution job on a selected computing node by initiating performance of the operations of the execution job that were not completed at the first time, the resumed execution including retrieving the persistently stored intermediate output data that was stored at the first time for the execution job and initiating storage at the second time of the retrieved output data on a portion of the distributed file system on the selected computing node; and

after the execution of the multiple execution jobs of the indicated program is completed, providing final results from the execution to the one user.

2. The method of claim 1 wherein, for each of at least one of the requests, the indicated program for the request is a program provided by the distributed program execution service that is designed to be separable in a defined manner into distinct execution jobs for parallel execution, the multiple execution jobs of the indicated program are automatically determined by separating the indicated program in the defined manner into a selected quantity of execution jobs, and the responding to the request further includes automatically separating the indicated input data for the request into the selected quantity of subsets and selecting one of the separated input data subsets for each of the determined execution jobs.

3. The method of claim 2 wherein the distributed program execution service is a fee-based service such that the users who supply the multiple received requests each pay a fee for the requested execution of the indicated program using the indicated input data, wherein the fee paid by one of the users corresponds to a low priority for the execution of the indicated program for the one user such that the execution of that indicated program is allowed to be temporarily suspended to execute other programs of higher priority, and wherein the determining to terminate the execution at the first time of the one or more execution jobs for the one user is based on temporarily suspending the execution of the indicated program for the one user by selecting all of the multiple execution jobs for the one user as the one or more execution jobs whose execution is determined to be terminated.

4. A computer-implemented method for managing distributed execution of programs, the method comprising:

under control of one or more computing systems that provide a distributed program execution service that manages distributed execution of programs for users, the distributed program execution service providing a plurality of computing nodes that are configurable to execute the programs for the users,

after execution of multiple execution jobs of an indicated program is initiated on multiple of the plurality of computing nodes, the execution of the indicated program being performed on behalf of a first user and using indicated input data in such a manner that the multiple execution jobs each have one or more operations to be performed using at least some of the indicated input data, automatically tracking information about a state of the execution of the multiple execution jobs on the multiple computing nodes, the tracking including identifying intermediate results that are produced from a subset of the operations of the multiple execution jobs whose performance is complete and that are stored on the multiple computing nodes;

after determining to terminate execution at a first time of at least one of the multiple execution jobs, the at least one execution jobs having at least one operation that is in the subset of operations whose performance is complete and having at least one other operation that is not in the subset and whose performance is not complete, automatically identifying the at least one operations by using the tracked information, and initiating persistent storage of the identified intermediate results produced from the at least one operations;

at a later second time after the first time, initiating a resumed execution of the at least one execution jobs on at least one computing node so as to complete the performance of the at least one other operations that are not in the subset whose performance is complete and so as to not repeat the completed performance of the at least one operations in the subset, the resumed execution being performed in a manner based at least in part on the persistently stored intermediate results; and

after the execution of the multiple execution jobs of the indicated program is completed, providing final results from the execution to the first user.

5. The method of claim 4 wherein the determining to terminate the execution at the first time of the at least one execution jobs is based at least in part on a determination to move execution of the at least one execution jobs from one or more computing nodes on which the at least one execution jobs are executing at the first time to one or more other computing nodes, wherein the one or more other computing nodes are the at least one computing nodes on which the resumed execution of the at least one execution jobs occurs at the second time, and wherein the second time occurs substantially immediately after the first time.

6. The method of claim 4 wherein the determining to terminate the execution at the first time of the at least one execution jobs is based at least in part on a request to temporarily suspend execution of the indicated program, and wherein the at least one execution jobs include all of the multiple execution jobs that are executing at the first time.

7. The method of claim 6 wherein the at least one computing node on which the resumed execution of the at least one execution jobs is initiated is distinct from one or more of the multiple computing nodes that executed the at least one execution jobs prior to the first time.

8. The method of claim 6 wherein the at least one computing node on which the resumed execution of the at least one execution jobs is initiated includes a selected quantity of one or more computing nodes that is distinct from a quantity of the multiple computing nodes.

9. The method of claim 6 wherein the request is received from the first user.

10. The method of claim 6 wherein the request is received from a second user distinct from the first user.

11. The method of claim 4 wherein the determining to terminate the execution at the first time of the at least one execution jobs is based at least in part on a determination to perform administrative activities at the first time on one or more computing nodes on which the at least one execution jobs are executing at the first time.

12. The method of claim 4 wherein the determining to terminate the execution at the first time of the at least one execution jobs is based at least in part on a determination to execute one or more other programs at the first time on one or more computing nodes on which the at least one execution jobs are executing at the first time, the one or more other programs having a higher priority than the indicated program.

13. The method of claim 4 further comprising determining that one of the multiple computing nodes on which one of the multiple execution jobs is executing has failed during the execution of the one execution job, and automatically re-starting at a later time execution of the one execution job on a distinct computing node, the re-started execution including repeating performance of at least one operation of the one execution job whose performance was previously completed before the one computing node failed.

14. The method of claim 4 wherein the execution of the multiple execution jobs on the multiple computing nodes is managed by software executing on a master node, wherein the automatic tracking of the state of the execution of the multiple execution jobs includes maintaining information on the master node about the state of the execution of the multiple execution jobs, and wherein the determining to terminate the execution at the first time of the at least one execution jobs is based at least in part on failure of the master node.

15. The method of claim 14 wherein the at least one execution jobs whose execution is terminated include all of the multiple execution jobs that are executing at the first time, wherein the initiating of the resumed execution of the at least one execution jobs includes initiating execution of a new second master node and restoring on the second master node the maintained information from the master node about the state of the execution of the multiple execution jobs, and wherein the resumed execution of the at least one execution jobs on the at least one computing node is managed by the second master node.

16. The method of claim 4 wherein the initiating of the persistent storage of the identified intermediate results includes interacting with a network-accessible remote storage service that persistently stores the identified intermediate results.

17. The method of claim 4 wherein the initiating of the persistent storage of the identified intermediate results includes storing the identified intermediate results at a location remote from the multiple computing nodes.

18. The method of claim 4 wherein the identified intermediate results stored on the multiple computing nodes are stored using a distributed file system that is implemented on the multiple computing nodes, and wherein the initiating of the persistent storage of the identified intermediate results includes creating one or more logical storage volumes to represent at least a portion of the distributed file system and to persistently store the identified intermediate results on one or more storage nodes remote from the multiple computing nodes.

19. The method of claim 4 further comprising, under the control of the one or more computing systems:

after the execution of one or more of the multiple execution jobs is completed, initiating execution of one or more additional execution jobs on one or more of the multiple computing nodes, at least one of the additional execution jobs using input data that includes output data generated by the completed execution of at least one of the one or more execution jobs; and

after the execution of the multiple execution jobs and of the additional execution jobs of the indicated program is completed, generating the final results to be provided to the first user, the generated final results being based at least in part on output data generated by completed execution of at least some of the multiple execution jobs and additional execution jobs.

20. The method of claim 4 wherein the execution of the multiple execution jobs of the indicated program is based on a request received from the first user to execute the indicated program using the indicated input data, wherein the received request from the first user includes one or more indications of additional configuration information for use in the executing of the indicated program that is used by the one or more computing systems as part of performing the execution of the indicated program, the additional configuration information including at least one of information about a quantity of the multiple computing nodes, of information about one or more criteria that computing nodes selected for use as the multiple computing nodes are to satisfy, of information about how to determine the multiple execution jobs of the indicated program, and of information about how to select subsets of the indicated input data to be used by the multiple execution jobs, wherein one or more other requests are received from one or more other users to execute one or more indicated programs using indicated other input data, and wherein the one or more computing systems further automatically respond to the other requests so as to perform distributed execution of the one or more indicated programs using the plurality of computing nodes.

21. The method of claim 20 wherein the additional configuration information includes information about the quantity of the multiple computing nodes and/or about one or more criteria that the computing nodes selected for use as the multiple computing nodes are to satisfy, and wherein the performing of the execution of the indicated program by the one or more computing systems includes automatically selecting the multiple computing nodes for use in performing a distributed execution of the indicated program based at least in part on the additional configuration information.

22. The method of claim 4 wherein the indicated program is designed to perform one or more map functions on each of multiple input data subsets and to perform one or more reduce functions on results of the one or more map functions, and wherein the method further comprises generating the multiple execution jobs to each implement at least one of the map functions and/or at least one of the reduce functions.

23. The method of claim 4 wherein the distributed program execution service uses virtualization technology such that the plurality of computing nodes include, for each of multiple physical computing systems, multiple virtual machines hosted by the physical computing system that are each able to execute at least one execution job, wherein the multiple computing nodes selected for the first user are each a hosted virtual machine, and wherein the multiple execution jobs are each executed as part of a virtual machine image.

24. A computer-readable medium whose contents enable a computing system to manage distributed execution of programs, by performing a method comprising:

tracking information about a state of distributed execution of an indicated program on multiple computing nodes, the indicated program having multiple execution jobs executing on the multiple computing nodes, the execution jobs each having one or more operations to be performed, the tracked state information including information about intermediate results produced from operations of the multiple execution jobs whose performance is complete;

after the execution of at least one of the execution jobs is terminated before performance of all of the operations of the at least one execution jobs is completed, the at least one execution jobs having at least one operation whose performance is complete, continuing the distributed execution of the indicated program by automatically initiating resumed performance of the operations of the at least one execution jobs other than the at least one operations; and

after the distributed execution of the indicated program is completed, providing an indication of final results from the distributed execution.

25. The computer-readable medium of claim 24 wherein the computing system is part of a distributed program execution service that provides a plurality of computing nodes configurable to perform distributed execution of programs for users of the distributed program execution service, wherein the distributed execution of the indicated program is initiated on behalf of a first user of the distributed program execution service and uses input data for the indicated program that is indicated by the first user, wherein the multiple computing nodes include a quantity of computing nodes specified by the first user and at least some of the multiple computing nodes are selected from the plurality of computing nodes, wherein the method further comprises determining at a first time to terminate the execution of the at least one execution jobs and automatically initiating persistent storage of the intermediate results produced from the at least one operation, and wherein the continuing of the distributed execution of the indicated program is performed at a second time later than the first time and includes retrieving and using at least some of the persistently stored intermediate results.

26. The computer-readable medium of claim 24 wherein the execution of the at least one execution jobs is terminated based on failure at a first time of one or more of the multiple computing nodes on which the at least one execution jobs were being executed, wherein the performance of the at least one operations of the at least one execution jobs includes persistently storing the intermediate results produced from completion of the performance at a time earlier than the first time, and wherein the continuing of the distributed execution of the indicated program is performed at a second time later than the first time and includes retrieving and using at least some of the persistently stored intermediate results.

27. The computer-readable medium of claim 24 wherein the computer-readable medium is at least one of a memory of a computing system that stores the contents and a data transmission medium that includes a generated stored data signal containing the contents, and wherein the contents are instructions that when executed cause the computing system to perform the method.

28. A computing system configured to manage distributed execution of programs, comprising:

one or more memories; and

a system manager component that is configured to manage distributed execution for users of a distributed execution service by, for each of multiple of the users;

receiving an indication from the user to perform distributed execution of multiple related execution jobs;

initiating execution of the multiple execution jobs on multiple computing nodes;

after a partial execution of at least one of the multiple execution jobs is performed but before the execution of the at least one execution jobs is completed, determining to terminate the execution of the at least one execution jobs, and automatically initiating persistent storage of an intermediate state of the partial execution of the at least one execution jobs;

at a later time, retrieving the persistently stored intermediate state of the partial execution of the at least one execution jobs, and resuming the execution of the at least one execution jobs based at least in part on the retrieved persistently stored intermediate state; and

after the execution of the multiple execution jobs is completed, providing final results from the execution to the user.

29. The computing system of claim 28 wherein the execution of the at least one execution jobs for one of the users uses a first group of input data and when completed produces a second group of output data, wherein the partial execution of the at least one execution jobs for the one user uses at least some of the first group of input data to produce a subset of the second group of output data, and wherein the persistent storage of the intermediate state of the partial execution of the at least one execution jobs for the one user includes storing the produced subset of the second group of output data for later use.

30. The computing system of claim 29 wherein the indication from the one user to perform the distributed execution of the multiple related execution jobs includes an indication of a program to be performed that is automatically separated by the distributed execution service into the multiple related execution jobs and includes an indication of input data to be used for the program that includes the first group of input data, wherein the at least one execution jobs of the one user include multiple operations that each operate on a portion of the first group of input data and when completed produce a portion of the second group of output data, wherein the partial execution of the at least one execution jobs of the one user includes a subset of the multiple operations being completed, wherein the produced subset of the second group of output data includes the portions of the second group of output data that are produced by the subset of the multiple operations that are completed, and wherein the persistent storage of the intermediate state of the partial execution of the at least one execution jobs for the one user further includes storing one or more portions of the first group of input data that were not used by the subset of the multiple operations that are completed to produce the subset of the second group of output data.

31. The computing system of claim 28 wherein the system manager component includes software instructions for execution by the computing system.

32. The computing system of claim 28 wherein the system manager component consists of a means for managing distributed execution for users of a distributed execution service by:

receiving an indication from the user to perform distributed execution of multiple related execution jobs;

initiating execution of the multiple execution jobs on multiple computing nodes;

after a partial execution of at least one of the multiple execution jobs is performed but before the execution of the at least one execution jobs is completed, determining to terminate the execution of the at least one execution jobs, and automatically initiating persistent storage of an intermediate state of the partial execution of the at least one execution jobs;

at a later time, retrieving the persistently stored intermediate state of the partial execution of the at least one execution jobs, and resuming the execution of the at least one execution jobs based at least in part on the retrieved persistently stored intermediate state; and

after the execution of the multiple execution jobs is completed, providing final results from the execution to the user.