Identifying nodes already storing indicated input data to perform distributed execution of an indicated program in a node cluster

• US 9,276,987 B1
• Filed: 03/11/2013
• Issued: 03/01/2016
• Est. Priority Date: 06/02/2009
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Sirota, Peter, Khanna, Richendra
• Primary Examiner(s): An, Meng
• Assistant Examiner(s): Joy-Davila, Jorge A Chu
• Application Number:
  US13/794,220
• Time in Patent Office: 1,086 Days
• Field of Search: None
• US Class Current: 1/1
• CPC Class Codes: G06F 9/5033 considering data affinity
• CPC Class Codes: H04L 67/00 Network arrangements or protocols for supporting network services or applications user-to-user messaging H04L51/00; network arrangements, protocols or services for supporting real-time applications in data packet communications networks H04L65/00

Abstract

Techniques are described for managing execution of programs, such as for distributed execution of a program on multiple computing nodes. In some situations, the techniques include selecting a cluster of computing nodes to use for executing a program based at least in part on data to be used during the program execution. For example, the computing node selection for a particular program may be performed so as to attempt to identify and use computing nodes that already locally store some or all of the input data that will be used by those computing nodes as part of the executing of that program on those nodes. Such techniques may provide benefits in a variety of situations, including when the size of input datasets to be used by a program are large, and the transferring of data to and/or from computing nodes may impose large delays and/or monetary costs.

Claims

1. A computer-implemented method comprising:

receiving, by one or more computing systems configured to provide a program execution service that executes programs for multiple users by using a plurality of computing nodes provided by the program execution service, configuration information regarding using indicated input data as part of executing an indicated program for a first user of the multiple users in a distributed manner on a computing node cluster;

selecting, by the one or more configured computing systems, multiple computing nodes from the plurality for the computing node cluster based at least in part on the selected multiple computing nodes each being identified as locally storing at least some of the indicated input data prior to the receiving of the configuration information;

determining, by the one or more configured computing systems and based on at least one of the selected multiple computing nodes being currently unavailable to perform the executing of the indicated program due to executing one or more other programs for one or more other users and based on the executing of the indicated program being determined to have a higher priority than the executing of the one or more other programs, to terminate the executing of the one or more other programs on the at least one selected computing node to enable use of the at least one selected computing node in the executing of the indicated program; and

initiating, by the one or more configured computing systems and based at least in part on the terminating of the executing of the one or more other programs on the at least one selected computing node, the executing of the indicated program on the selected multiple computing nodes using the locally stored at least some indicated input data.

2. The method of claim 1 wherein the initiating of the executing of the indicated program on the selected multiple computing nodes includes, for each of the multiple computing nodes, initiating use of the at least some indicated input data that is locally stored on the computing node as input for the executing of the indicated program on the computing node.

3. The method of claim 1 wherein the receiving of the configuration information includes receiving a request from the first user to perform the executing of the indicated program and includes receiving access to the indicated input data.

4. The method of claim 1 wherein the selecting of the multiple computing nodes for the computing node cluster includes selecting an additional computing node for the computing node cluster that does not already locally store any of the indicated input data prior to the receiving of the configuration information, and initiating local storing, on a local storage device of the additional computing node, of some or all of the indicated input data for use as part of the executing of the indicated program on the additional computing node.

5. The method of claim 4 wherein the initiating of the local storing of the some or all indicated input data on the local storage device includes copying the some or all indicated input data from at least one storage node remote from the selected multiple computing nodes.

6. The method of claim 4 wherein the initiating of the local storing of the some or all indicated input data on the local storage device of the additional computing node includes attaching a non-local data storage volume to the additional computing node and copying the some or all indicated input data from the attached non-local data storage volume to the local storage device, the non-local data storage volume being provided by a remote storage service.

7. The method of claim 1 wherein the initiating of the executing of the indicated program on the selected multiple computing nodes includes, for an additional computing node that locally stores some or all of the indicated input data and that is executing another program, delaying use of the additional computing node for the computing node cluster until the executing of the another program has ended.

8. The method of claim 1 wherein executing of the indicated program begins and is temporarily suspended before the selecting of the multiple computing nodes, and wherein the selecting of the multiple computing nodes is performed as part of resuming the temporarily suspended execution of the indicated program.

9. The method of claim 1 wherein the received configuration information further includes one or more indicated criteria that the selected multiple computing nodes are to satisfy, and wherein the selecting of the multiple computing nodes is further performed to satisfy the one or more indicated criteria.

10. The method of claim 1 further comprising identifying that one or more additional computing nodes are available at a future time to participate in the computing node cluster for performing the executing of the indicated program, and initiating use of the one or more additional computing nodes in the computing node cluster at the future time.

11. The method of claim 1 wherein the program execution service provides each of the selected multiple computing nodes as a virtual machine hosted by a physical computing system.

12. A non-transitory computer-readable medium having stored contents that configure a computing system to:

receive information that indicates a program to be executed for a first user by a program execution service on multiple computing nodes using indicated input data;

identify, by the configured computing system, one or more computing nodes for executing the indicated program based at least in part on the one or more computing nodes already storing at least some of the indicated input data and being available for performing the executing of the indicated program, the identified one or more computing nodes being selected from a plurality of computing nodes used by the program execution service for executing programs;

initiate, by the configured computing system, the executing of the indicated program on the identified one or more computing nodes so as to use the already stored at least some indicated input data; and

initiate, by the configured computing system, the executing of the indicated program on an additional computing node that is identified as already storing at least some of the indicated input data, the initiating of the executing of the indicated program on the additional computing node including terminating execution of another program on the additional computing node for another user to enable the additional computing node to be available to perform the executing of the indicated program.

13. The non-transitory computer-readable medium of claim 12 wherein the received information is configuration information from the first user that includes an indication of a quantity of the multiple computing nodes, and wherein the executing of the indicated program is performed on a cluster of multiple computing nodes of the indicated quantity that include the identified one or more computing nodes and the additional computing node.

14. The non-transitory computer-readable medium of claim 12 wherein the configured computing system is part of the program execution service, wherein the first user is one of multiple users of the program execution service on whose behalf multiple programs are executed, and wherein the stored contents include executable software instructions.

15. The non-transitory computer-readable medium of claim 12 wherein the receiving of the information includes receiving a request for the program execution service to execute the indicated program, wherein the identifying of the one or more computing nodes is further based at least in part on the one or more computing nodes storing the at least some indicated input data before the receiving of the request, and wherein the stored contents further configure the computing system to select a further computing node for use in the executing the indicated program and to initiate local storing, on one or more local storage devices of the selected further computing node, of some or all of the indicated input data for use as part of the executing of the indicated program on the selected further computing node.

16. The non-transitory computer-readable medium of claim 12 wherein executing of the indicated program begins and is temporarily suspended before identifying of the one or more computing nodes, and wherein the identifying of the one or more computing nodes is performed as part of resuming the temporarily suspended execution of the indicated program and includes selecting at least one computing node that was used for the executing of the indicated program before the executing was temporarily suspended.

17. The non-transitory computer-readable medium of claim 12 wherein the stored contents further configure the computing system to determine that the identified one or more computing nodes are available for performing the executing of the indicated program at a future time, and wherein the initiating of the executing of the indicated program on the identified one or more computing nodes is performed at the future time.

18. A system comprising:

one or more processors of one or more computing systems; and

one or more components of a program execution service that, when executed by at least one of the one or more processors, cause the at least one processor to;

receive a request to execute, for a first user that is one of multiple users for which the program execution service executes programs by using a plurality of computing nodes provided by the program execution service, an indicated program using indicated input data;

select multiple computing nodes from the plurality to use for executing the indicated program based on the multiple computing nodes already storing at least some of the indicated input data before the receiving of the request;

determine that at least one of the selected multiple computing nodes is currently unavailable to perform the executing of the indicated program due to executing one or more other programs for one or more other users, that the executing of the indicated program on the at least one selected computing node is preferred for the program execution service, and to terminate the executing of the one or more other programs on the at least one selected computing node to enable use of the at least one selected computing node in the executing of the indicated program; and

initiate, based at least in part on the terminating of the executing of the one or more other programs on the at least one selected computing node, the executing of the indicated program on the multiple computing nodes to cause to cause use of the already stored at least some indicated input data during the executing of the indicated program.

19. The system of claim 18 wherein the one or more components include software instructions that further configure the one or more components to provision the selected multiple computing nodes to prepare the provisioned selected multiple computing nodes to perform the executing of the indicated program.

20. The system of claim 18 wherein the received request includes configuration information indicated by the first user.

21. The system of claim 18 wherein the selected multiple computing nodes further include one or more additional computing nodes that do not already store at least some of the indicated input data, and wherein the one or more components are further configured to initiate local storing, on one or more local storage devices of the one or more additional computing nodes, of some or all of the indicated input data for use as part of the executing of the indicated program on the one or more additional computing nodes.

22. The system of claim 18 wherein executing of the indicated program begins and is temporarily suspended before the selecting of the multiple computing nodes, and wherein the initiating of the executing of the indicated program on the selected multiple computing nodes is performed as part of resuming the temporarily suspended execution of the indicated program.