Saving program execution state

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.

1 - 32 . (canceled) 
     
     
         33 . A computer-implemented method comprising:
 initiating, by one or more configured computing systems, distributed execution of multiple execution jobs of an indicated program on multiple computing nodes;   persistently storing, by the one or more configured computing systems, state information from a partial completion at a first time of the distributed execution, the stored state information including intermediate results produced from one or more first execution jobs that are completed at the first time and not including intermediate results from one or more other second execution jobs that are not completed at the first time;   continuing, by the one or more configured computing systems and after the first time, the distributed execution of the indicated program by initiating resumed performance of the one or more second execution jobs without further performing the one or more first execution jobs, the initiating of the resumed performance including using at least some of the intermediate results from the stored state information as input for at least one of the second execution jobs; and   providing an indication of final results from the distributed execution of the indicated program after the distributed execution is completed.   
     
     
         34 . The computer-implemented method of  claim 33  wherein the one or more configured computing systems are part of a distributed program execution service that provides a plurality of computing nodes configurable to execute programs for users of the distributed program execution service, wherein the distributed execution of the indicated program is initiated on behalf of a user of the distributed program execution service, and wherein the multiple computing nodes are selected from the plurality of computing nodes. 
     
     
         35 . The computer-implemented method of  claim 33  wherein the persistent storing of the state information is initiated based at least in part on failure of one or more of the multiple computing nodes during the distributed execution, and wherein the initiating resumed performance of the one or more second execution jobs includes executing the one or more second execution jobs on at least one remaining computing node of the multiple computing nodes that is distinct from the one or more computing nodes. 
     
     
         36 . The computer-implemented method of  claim 33  wherein the persistent storing of the state information is initiated based at least in part on failure of one or more of the multiple computing nodes during the distributed execution, and wherein the initiating resumed performance of the one or more second execution jobs includes adding one or more additional computing nodes to the multiple computing nodes to replace the failed one or more computing nodes, and includes executing the one or more second execution jobs at least in part on the one or more additional computing nodes. 
     
     
         37 . The computer-implemented method of  claim 33  wherein execution of the one or more second execution jobs is partially completed at the first time, wherein the method further comprises determining to terminate the partially completed execution at the first time of the one or more second execution jobs to enable moving the execution of the one or more second execution jobs from one or more computing nodes on which the one or more second execution jobs are executing before the first time to one or more other computing nodes, and wherein the initiating resumed performance of the one or more second execution jobs includes executing the one or more second execution jobs on the one or more other computing nodes. 
     
     
         38 . The computer-implemented method of  claim 33  further comprising determining to terminate the distributed execution at the first time based at least in part on a received request from a user to temporarily suspend execution of the indicated program, and terminating, in response to the request, execution of any of the multiple execution jobs that are executing at the first time. 
     
     
         39 . The computer-implemented method of  claim 38  wherein the continuing of the distributed execution of the indicated program is performed in response to a further determination to resume the distributed execution that occurs after the terminating of the execution, and wherein a second quantity of computing nodes on which the continued distributed execution occurs is distinct from a first quantity of the multiple computing nodes at a time of the terminating of the execution. 
     
     
         40 . The computer-implemented method of  claim 33  further comprising determining to terminate the distributed execution at the first time based at least in part on a determination to perform administrative activities on one or more computing nodes on which the distributed execution is occurring at the first time. 
     
     
         41 . The computer-implemented method of  claim 33  further comprising determining to terminate the distributed execution at the first time 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 distributed execution is occurring at the first time, the one or more other programs having a higher priority than the indicated program. 
     
     
         42 . The computer-implemented method of  claim 33  wherein the distributed execution of the indicated program on the multiple computing nodes is managed by software executing on a master node, and wherein the method further comprises maintaining information on the master node before the first time about a state of the distributed execution, and determining to terminate the execution at the first time based at least in part on failure of the master node. 
     
     
         43 . The computer-implemented method of  claim 42  wherein the persistent storing of the state information includes persistently storing the maintained information from the master node, wherein the continuing of the distributed execution includes initiating execution of a second master node and restoring, on the second master node, the maintained information from the master node, and wherein the resumed performance is managed by the second master node. 
     
     
         44 . The computer-implemented method of  claim 33  wherein the persistent storing of the state information includes providing instructions over one or more computer networks to a network-accessible remote storage service to cause persistent storage of the state information by the remote storage service. 
     
     
         45 . The computer-implemented method of  claim 33  further comprising receiving, by the one or more configured computing systems, configuration information from a user for use in the distributed execution of the indicated program, the configuration information indicating at least one of a group of types of information that include information about a quantity of the multiple computing nodes, information about one or more criteria that computing nodes selected for use as the multiple computing nodes are to satisfy, information about how to determine the multiple execution jobs of the indicated program, or information about how to select subsets of indicated input data to be used by the multiple execution jobs. 
     
     
         46 . The computer-implemented method of  claim 45  wherein the configuration information indicates information about the quantity of the multiple computing nodes and about the one or more criteria, and wherein the method further comprises selecting, by the one or more configured computing systems, the multiple computing nodes for use in performing the distributed execution of the indicated program based at least in part on the configuration information. 
     
     
         47 . The computer-implemented method of  claim 33  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 map function or at least one reduce function, the one or more first execution jobs implementing at least one map function and the one or more second execution jobs implementing at least one reduce function. 
     
     
         48 . A non-transitory computer-readable storage medium having stored contents that configure a computing system to:
 initiate, by the configured computing system, distributed execution of multiple execution jobs of an indicated program on multiple computing nodes;   persistently store state information from a partial completion at a first time of the distributed execution, the stored state information including intermediate results produced from one or more first execution jobs that are completed at the first time and not including intermediate results from one or more other second execution jobs that are not completed at the first time;   continue, by the configured computing system and after the first time, the distributed execution of the indicated program by initiating resumed performance of the one or more second execution jobs without further performing any of the one or more first execution jobs, wherein the resumed performance of at least one of the second execution jobs uses at least some of the intermediate results from the stored state information; and   provide an indication of final results from completion of the distributed execution of the indicated program.   
     
     
         49 . The non-transitory computer-readable storage medium of  claim 48  wherein execution of the one or more second execution jobs is partially completed at the first time, and wherein the stored contents include software instructions that, when executed, program the computing system to further determine to terminate the partially completed execution of the one or more second execution jobs at the first time and to initiate the persistent storing of the state information based at least in part on the determining to terminate the partially completed execution. 
     
     
         50 . The non-transitory computer-readable storage medium of  claim 49  wherein the one or more second execution jobs are executing on one or more computing nodes at the first time, and wherein the determining to terminate the partially completed execution of the one or more second execution jobs is performed to enable moving the execution of the one or more second execution jobs to one or more other computing nodes, to respond to a received request from a user to temporarily suspend execution of the indicated program, to perform administrative activities on the one or more computing nodes, or to execute one or more other programs with a higher priority on the one or more computing nodes. 
     
     
         51 . A system comprising:
 one or more hardware processors of one or more computing systems; and   one or more modules that, when executed by at least one of the one or more hardware processors, configure the at least one hardware processor to:
 initiate distributed execution of multiple execution jobs of an indicated program on multiple computing nodes; 
 persistently store state information at a first time from a partial completion of the distributed execution, the stored state information including intermediate results produced from one or more first execution jobs that are completed at the first time and not including intermediate results from one or more other second execution jobs that are not completed at the first time; 
 continue the distributed execution of the indicated program after the first time by initiating further performance of the one or more second execution jobs without further performing any of the one or more first execution jobs, wherein the further performance of at least one of the second execution jobs uses at least some of the intermediate results from the stored state information; and 
 provide an indication of final results from completion of the distributed execution of the indicated program. 
   
     
     
         52 . The system of  claim 51  wherein execution of the one or more second execution jobs is partially completed at the first time, and wherein the one or more modules include software instructions that, when executed, program the one or more computing systems to further determine at the first time to terminate the partially completed execution of the one or more second execution jobs and to initiate the persistent storing of the state information based at least in part on the terminating.