Executing threads of an application across multiple computing devices in a distributed virtual machine environment

According to one embodiment, a computer-implemented method includes executing code for an application using a computing resource of a first computing device. The application requests execution of a first thread and a second thread. The first thread is executed using the computing resource of the first computing device. A second computing device is selected from a plurality of computing devices. The second computing device has an available computing resource to execute the second thread. The second thread is assigned to the second computing device. The second computing device is operable to execute the second thread using the available computing resource.

BACKGROUND

A virtual machine is a software implementation of a machine (i.e. a computer) that executes applications like a computing device. A virtual machine may behave like a physical computer and contain its own virtual (i.e., software-based) CPU, RAM hard disk and network interface card (NIC). In some situations, a virtual machine may be provided with portions of physical hardware. For example, a virtual machine may be allocated portions of memory, processing time on a physical processor, or bandwidth on a physical network interface card. In some contexts, a virtual machine may be called a “guest” while the computing device, within which the virtual machine runs, may be called a “host.”

DESCRIPTION OF EXAMPLE EMBODIMENTS

As described above, a virtual machine is a software implementation of a machine (i.e. a computer) that executes applications like a computing device. A virtual machine may execute applications by using computing resources of its host computing device. Computing resources may include any hardware, firmware, and software resources, including, but not limited to, processors, memory, storage, and input-output devices. For example, a virtual machine may execute an application by assigning application threads to processors of the host computing device. A thread is a unit of processing that may be assigned or scheduled by an application or operating system.

As an application grows in size, the application may use additional resources from the host computing device. For example, larger applications may include additional threads to be executed on additional processors. To meet the increased demands, host computing devices may be provided with additional computing resources such as additional processors. Computing devices having additional resources, however, are more expensive to purchase and manage. Accordingly, some embodiments provide, among other capabilities, a distributed virtual machine environment across multiple computing devices.

For example, some embodiments provide a distributed virtual machine environment to execute an application across two or more computing devices. In some embodiments, this distributed virtual environment is transparent to the application and to users of the application. For example, the application may be written to be executed by one computing device, and the distributed virtual environment may execute the application across multiple computing devices with minimal or no changes to the application code. In this example, the application may not be aware that the distributed virtual machine environment is executing the application across multiple computing devices.

Some embodiments may execute an application across two or more computing devices with minimal latency. Generally, data travels faster between two processors within the same computing device than between two processors associated with different computing devices. Some embodiments may increase data transfer speeds between two computing devices by providing a high-speed network connection between the two computing devices. For example, a gigabit Ethernet connection may be provided between two computing devices hosting a distributed virtual machine environment. In some embodiments, the gigabit Ethernet connection may support data transfer rates on the order of 10 or 100 gigabits per second.

FIG. 1illustrates an example system100operable to provide a distributed virtual machine environment across two or more computing devices. In the illustrated example, system100includes a client system110, a machine manager130, computing devices140, and a shared repository160, which may communicate using network120and/or network connections125. Although system100is illustrated and primarily described as including particular components, the present disclosure contemplates system100including any suitable components, according to particular needs.

In general, system100is operable to execute application150across multiple computing devices140using a distributed virtual machine environment. In the illustrated example, application150has four threads152,154,156, and158. Computing devices140are operable to implement virtual machines142. In the illustrated example, computing device140aimplements virtual machine142a, computing device140bimplements virtual machine142b, computing device140cimplements virtual machine142c, and computing device140nimplements virtual machine142n. Each virtual machine142may be operable to execute a sub-instance of application150. In the illustrated example, virtual machine142aexecutes application sub-instance150a, and virtual machine142bexecutes application sub-instances150band150c. Each application sub-instance150is operable to execute threads of application150. In this example, application sub-instance150ainstructs computing device140ato execute threads152and154, application sub-instance150binstructs computing device140bto execute threads156, and application sub-instance150cinstructs computing device140bto execute threads158.

Client110may be implemented using any suitable type of processing system and may include any suitable combination of hardware, firmware, and software. One example of a machine client110may include, but is not limited to, computer system400ofFIG. 4. In the illustrated example, client110includes local storage112and local devices114. Local storage112may include any storage device associated with client110, including, but not limited to, an internal or external hard drive. Local devices114may include any input/output devices associated with client110, including, but not limited to, a monitor/screen, a mouse, a keyboard, a printer, or a scanner.

In general, client110may include one or more computer systems at one or more locations. Each computer system may include any appropriate input devices, output devices, mass storage media, processors, memory, or other suitable components for receiving, processing, storing, and communicating data. For example, each computer system may include a personal computer, workstation, network computer, kiosk, wireless data port, personal data assistant (PDA), one or more Internet Protocol (IP) telephones, smart phones, table computers, one or more servers, a server pool, one or more processors within these or other devices, or any other suitable processing device. Client110may be a stand-alone computer or may be a part of a larger network of computers associated with an entity.

Network120may represent any number and combination of wireline and/or wireless networks suitable for data transmission, including, but not limited to, network442ofFIG. 4. In some embodiments, network120may include an internet connection between client110and machine manager130.

Network connections125may represent any number and combination of wireline and/or wireless network connections suitable for data transmission, including, but not limited to, network442ofFIG. 4. In some embodiments, network connections125may include high-speed network connections among computing devices140and repository160. For example, a gigabit Ethernet connection may be provided between two computing devices140hosting a distributed virtual machine environment. In some embodiments, the gigabit Ethernet connection may support data transfer rates on the order of 10 or 100 gigabits per second.

Machine manager130may include any suitable type of processing system and may be implemented using any suitable combination of hardware, firmware, and software. One example of a machine manager130may include, but is not limited to, computer system400ofFIG. 4. In general, machine manager130may be operable to manage computing devices140and route communications between client110and computing devices140. For example, in one example embodiment, client110may request execution of application150. In this example embodiment, machine manager130may select computing device140afrom among the computing devices140and route communications between client110and computing device140a. In another example embodiment, machine manager130may monitor workloads of each computing device140. For example, machine manager130may be operable to identify computing devices140with computing resources, such as processors and memory, available to execute threads of application150. Machine manager130may also balance loads among computing devices140. For example, machine manager130may instruct virtual machine142ato relocate threads152and154to another computing device if computing device140ais running low on available resources.

Each computing device140may be implemented using any suitable type of processing system and may include any suitable combination of hardware, firmware, and software. One example of a computing device140may include, but is not limited to, computer system400ofFIG. 4. In the illustrated example, four computing devices140a,140b,140c, and140nare shown, although embodiments of system100may include more or fewer computing devices140.

Computing devices140may reside in any suitable location. In one example embodiment, at least some of computing devices140share space in the same computing rack or server farm. In some embodiments, computing devices140may be located in different geographic areas.

Each computing device140is operable to implement a virtual machine142. In the illustrated example, computing device140aimplements virtual machine142a, computing device140bimplements virtual machine142b, computing device140cimplements virtual machine142c, and computing device140nimplements virtual machine142n. In general, virtual machine142may represent any software implementation of a machine (i.e. a computer) that executes applications like a computing device. Virtual machine142may execute applications by using resources of its host computing device. For example, virtual machine142amay execute threads152and154using processors of computing device140a.

Each computing device140may include local storage that is accessible by a virtual machine142. In the illustrated example, computing device140aincludes local storage144aaccessible by virtual machine142a, and computing device140bincludes local storage144baccessible by virtual machine142b. One example of a local storage144may include, but is not limited to, storage module422ofFIG. 4.

Application150may include any collection of programs, procedures, algorithms, and instructions to be executed. In the illustrated example, application150includes instructions to execute four threads152,154,156, and158. Threads152,154,156, and158may represent any unit of processing that may be assigned or scheduled by an application or operating system. In the illustrated example, threads152and154are assigned to computing device140a, and threads156and158are assigned to computing device140b. A thread may be assigned to a computing device140by identifying any component or feature of the computing device140, including, but not limited to, identifying a name of the computing device, a processor of the computing device, a virtual machine running on the computing device, and/or identifying application sub-instances that happen to be associated with a particular computing device. In addition, threads may be assigned by any entity associated with application150. In the example ofFIG. 1, threads may be assigned by computing device140a, virtual machine142a, and/or application sub-instance150a.

Each virtual machine142may be operable to execute a sub-instance of application150. Each sub-instance of application150includes the programs, procedures, algorithms, and instructions for executing threads of application150. A sub-instance may not necessarily include all of the code for application150. For example, in some embodiments, a sub-instance may only include code relevant to or otherwise appropriate for execution of a particular thread.

In the illustrated example, virtual machine142aexecutes application sub-instance150a, and virtual machine142bexecutes application sub-instances150band150c. Each application sub-instance is operable to execute threads of application150. In this example, application sub-instance150ainstructs computing device140ato execute threads152and154, application sub-instance150binstructs computing device140bto execute threads156, and application sub-instance150cinstructs computing device140bto execute threads158.

Shared repository160may represent any storage device accessible by two or more computing devices140. One example of a shared repository160may include, but is not limited to, storage module406ofFIG. 4. In some embodiments, application code150may be stored on shared repository160in place of or in addition to being stored on computing device140a.

In some embodiments, shared repository160may store state information162. In general, a state is a unique configuration of information in a program or machine. State information162may include properties that may be modified during execution of a thread of application150. In one example embodiment, shared repository160includes state information162that may be modified by thread152of computing device140aand by thread156of computing device140b.

In some embodiments, state information162may be checked out to one thread, which may temporarily prohibit other threads from modifying state information162. Thus, in some embodiments, state information162may be locked such that other threads do not modify the state information. For example, if execution of thread152will result in changes to state information162, state information162may be checked out to thread152, preventing threads154,156, and158from modifying state information162.

In some embodiments, all or some of state information162may be stored on computing devices140either temporarily or permanently. For example, if execution of thread152will result in changes to state information162, state information162may be checked out and copied to local storage144a. In this example, thread152may modify the copy of state information162on local storage144a. This modified copy may be communicated back to shared storage160. The modified copy may replace the checked out state information162, and state information162may be unlocked.

In some embodiments, shared repository160may store read-only information164. Read-only information164may include any information accessed by a thread but that is not modifiable by the thread. Read-only information164may also be stored locally by the computing devices140(e.g., in local storage144). For example, if computing device140cinitiates virtual machine142c, virtual machine142cmay download a local copy of read-only information164to computing device140c. Storing read-only information164locally may improve the speed of thread execution.

In operation of an example embodiment of system100, virtual machines142provide a distributed virtual machine environment in which to execute application150across multiple computing devices140. In one example, client110requests execution of application150. Machine manager130routes the request from client110to computing device140a. Computing device140aimplements virtual machine142aand executes application150within virtual machine142a.

Application150may request execution of multiple threads. In this example, application150requests execution of four threads152,154,156, and158. Virtual machine142aassigns the threads to available computing resources. In this example, virtual machine142aassigns the four threads152,154,156, and158to four processors. Virtual machine142aassigns threads152and154to processors local to computing device140aand assigns threads156and158to processors associated with computing device140b, which is remote from computing device140a. In this example, application150is unaware that threads156and158will be executed on computing device140b. Rather, the existence and location of computing device140bmay be known to virtual machine142abut not application150.

Virtual machine142ainitiates application sub-instance150ato execute threads152and154. Application sub-instance150aincludes the code from application150which may be used to execute threads152and154. Virtual machine142ainstructs computing device140bto initiate application sub-instances150band150c. Application sub-instance150bincludes the code from application150which may be used to execute thread156on computing device140b. Application sub-instance150cincludes the code from application150which may be used to execute thread158on computing device140b.

In the example ofFIG. 1, virtual machines142aand142bcommunicate across network connection125. In one example embodiment, network connection125may be a high-speed network connection, such as a gigabit network connection. In some embodiments, providing a high-speed network connection between virtual machines142aand142bmay allow virtual machines142aand142bto provide a distributed virtual machine environment that executes threads quickly regardless of the physical location of the computing resources.

In the example ofFIG. 1, virtual machine142aassigns threads152and154to computing resources local to computing device140a. In some embodiments, however, virtual machine142amay not assign any threads to computing resources local to computing device140a. Rather, virtual machine142amay assign all threads to remote computing resources. In this example, virtual machine142amay act as a master virtual machine, managing and monitoring the execution of threads across remote computing resources. For example, virtual machine142amay maintain a table of threads that identifies where each thread is assigned even if none of the threads are assigned to computing device140a.

In the example ofFIG. 1, virtual machine142aassigns threads156and158to computing device140b. In one example circumstance, however, computing device140bmay suffer a failure and be unable to execute threads156and158. In this example embodiment, virtual machine142amay detect failure of computing device140band assign threads156and158to one or more other computing devices (e.g., computing device140c). Virtual machine142amay detect failure of computing device140b, for example, if computing device140bdoes not execute threads156and158within an expected time period.

FIG. 2shows an example method200for assigning threads in a distributed virtual machine environment implemented across two or more computing devices. At step210, computing device140executes application150in virtual machine142a. In this example, application150requests execution of four threads152,154,156, and158. In this example, application150also requires certain computing resources for execution of threads152,154,156, and158. For example, application150may require that any computing device have certain available processing capability, available memory, available storage, or available input-out devices. Steps220through260address an example method for assigning thread156to computing device having the required computing resources.

At step220, virtual machine142aidentifies a plurality of computing devices140including computing devices140b,140c, and140n. At step230, virtual machine142adetermines whether computing device140bhas the required computing resources for execution of thread156. For example, application150may require that the computing device have an available processor, and virtual machine142amay determine at step230whether computing device140bhas an available processor for executing thread156. If not, virtual machine142adetermines at step240whether other computing devices140, such as computing devices140cand140n, have an available processing unit. In this example, however, computing device140bdoes have an available processing unit for executing thread156. Thus, at step250, virtual machine142aassigns thread156to computing device140b.

At step260, virtual machine142bexecutes thread156on computing device140bto yield thread output. Thread output may include any information received as a result of executing a thread. For example, thread output may include updated state information162. In this example, execution of thread156may include providing a mechanism to update state information162based on the thread output. As one example, execution of thread156may include identifying state information162associated with thread156and locking the identified state information162such that computing device140ais prevented from modifying state information162. In this example embodiment, virtual machine142amay transmit a copy of state information162to computing device140b. Virtual machine142bmay then instruct computing device140bto execute thread156and receive the thread output. Virtual machine142bmay then update state information162with the thread output and then unlock state information162. Unlocking state information162may allow other elements or processes to again modify state information162.

FIG. 3illustrates an example method for updating state information shared by virtual machines, according to some embodiments. In this example, execution of thread156yields updates to state information162.

At step310, virtual machine142bidentifies state information162associated with thread156. In this example, state information162associated with thread156includes state information162that will be updated as a result of execution of thread156. At step320, virtual machine142blocks the state information162associated with thread156. Locking the state information162associated with thread156may prevent threads152,154, and158from modifying the state information162associated with thread156.

At step330, virtual machine142bcopies the state information162associated with thread156to local storage144b. At step340, virtual machine142bexecutes thread156and receives thread output as a result of execution of thread156. At step350, virtual machine142bupdates the state information162associated with thread156stored on shared storage160. At step360, virtual machine142bunlocks the state information162associated with thread156.

FIG. 4illustrates an example computer system400that may be used for one or more sub-instances of example system100ofFIG. 1, according to some embodiments. Although the present disclosure describes and illustrates a particular computer system400having particular components in a particular configuration, the present disclosure contemplates any suitable computer system having any suitable components in any suitable configuration. Moreover, computer system400may have take any suitable physical form, such as for example one or more integrated circuit (ICs), one or more printed circuit boards (PCBs), one or more handheld or other devices (such as mobile telephones or PDAs), one or more personal computers, one or more super computers, one or more servers, and one or more distributed computing elements. Portions or all of client110, processing system204, storage module108, and computing resources110may be implemented using all of the components, or any appropriate combination of the components, of computer system400described below.

Computer system400may have one or more input devices402(which may include a keypad, keyboard, mouse, stylus, or other input devices), one or more output devices404(which may include one or more displays, one or more speakers, one or more printers, or other output devices), one or more storage devices406, and one or more storage media408. An input device402may be external or internal to computer system400. An output device404may be external or internal to computer system400. A storage device406may be external or internal to computer system400. A storage medium408may be external or internal to computer system400.

System bus410couples subsystems of computer system400to each other. Herein, reference to a bus encompasses one or more digital signal lines serving a common function. The present disclosure contemplates any suitable system bus410including any suitable bus structures (such as one or more memory buses, one or more peripheral buses, one or more a local buses, or a combination of the foregoing) having any suitable bus architectures. Example bus architectures include, but are not limited to, Industry Standard Architecture (ISA) bus, Enhanced ISA (EISA) bus, Micro Channel Architecture (MCA) bus, Video Electronics Standards Association local (VLB) bus, Peripheral Component Interconnect (PCI) bus, PCI-Express bus (PCI-X), and Accelerated Graphics Port (AGP) bus.

Computer system400includes one or more processors412(or central processing units (CPUs)). A processor412may contain a cache414for temporary local storage of instructions, data, or computer addresses. Processors412are coupled to one or more storage devices, including memory416. Memory416may include RAM418and ROM420. Data and instructions may transfer bi-directionally between processors412and RAM418. Data and instructions may transfer uni-directionally to processors412from ROM420. RAM418and ROM420may include any suitable computer-readable storage media.

Computer system400includes fixed storage422coupled bi-directionally to processors412. Fixed storage422may be coupled to processors412via storage control unit407. Fixed storage422may provide additional data storage capacity and may include any suitable computer-readable storage media. Fixed storage422may store an operating system (OS)424, one or more executables (EXECs)426, one or more applications or programs428, data430and the like. Fixed storage422is typically a secondary storage medium (such as a hard disk) that is slower than primary storage. In appropriate cases, the information stored by fixed storage422may be incorporated as virtual memory into memory416. In some embodiments, fixed storage422may include network resources, such as one or more storage area networks (SAN) or network-attached storage (NAS). In some embodiments, operating system424, EXECs426, application or programs428, and data430may be stored on storage device406and/or storage medium408instead of or in addition to storage422.

Processors412may be coupled to a variety of interfaces, such as, for example, graphics control432, video interface434, input interface436, output interface437, and storage interface438, which in turn may be respectively coupled to appropriate devices. Example input or output devices include, but are not limited to, video displays, track balls, mice, keyboards, microphones, touch-sensitive displays, transducer card readers, magnetic or paper tape readers, tablets, styli, voice or handwriting recognizers, biometrics readers, or computer systems. Network interface440may couple processors412to another computer system or to network442. Network interface440may include wired, wireless, or any combination of wired and wireless components. Such components may include wired network cards, wireless network cards, radios, antennas, cables, or any other appropriate components. With network interface440, processors412may receive or send information from or to network442in the course of performing steps of some embodiments. Some embodiments may execute solely on processors412. Some embodiments may execute on processors412and on one or more remote processors operating together.

In a network environment, where computer system400is connected to network442, computer system400may communicate with other devices connected to network442. Computer system400may communicate with network442via network interface440. For example, computer system400may receive information (such as a request or a response from another device) from network442in the form of one or more incoming packets at network interface440and memory416may store the incoming packets for subsequent processing. Computer system400may send information (such as a request or a response to another device) to network442in the form of one or more outgoing packets from network interface440, which memory416may store prior to being sent. Processors412may access an incoming or outgoing packet in memory416to process it, according to particular needs.

Network442may represent any number and combination of wireline and/or wireless networks suitable for data transmission, including, but not limited to, network120ofFIG. 1. Network442may, for example, communicate internet protocol packets, frame relay frames, asynchronous transfer mode cells, and/or other suitable data between network addresses. Network442may include a public or private data network; one or more intranets; a local area network (LAN); a metropolitan area network (MAN); a wide area network (WAN); a wireline or wireless network; a local, regional, or global communication network; an optical network; a satellite network; a cellular network; an enterprise intranet; all or a portion of the Internet; other suitable communication links; or any combination of the preceding. Although the illustrated embodiment shows one network442, teachings of some embodiments recognize that more or fewer networks may be used and that not all elements may communicate via a network. Teachings of some embodiments also recognize that communications over a network is one example of a mechanism for communicating between parties, and any suitable mechanism may be used.

Some embodiments involve one or more computer-storage products that include one or more tangible, computer-readable storage media that embody software for performing one or more steps of one or more processes described or illustrated herein. In some embodiments, one or more portions of the media, the software, or both may be designed and manufactured specifically to perform one or more steps of one or more processes described or illustrated herein. Additionally or alternatively, one or more portions of the media, the software, or both may be generally available without design or manufacture specific to processes described or illustrated herein. Example computer-readable storage media include, but are not limited to, CDs (such as CD-ROMs), FPGAs, floppy disks, optical disks, hard disks, holographic storage devices, ICs (such as ASICs), magnetic tape, caches, PLDs, RAM devices, ROM devices, semiconductor memory devices, and other suitable computer-readable storage media. In some embodiments, software may be machine code which a compiler may generate or one or more files containing higher-level code which a computer may execute using an interpreter.

As an example and not by way of limitation, memory416may include one or more tangible, computer-readable storage media embodying software and computer system400may provide particular functionality described or illustrated herein as a result of processors412executing the software. Memory416may store and processors412may execute the software. Memory416may read the software from the computer-readable storage media in mass storage device416embodying the software or from one or more other sources via network interface440. When executing the software, processors412may perform one or more steps of one or more processes described or illustrated herein, which may include defining one or more data structures for storage in memory416and modifying one or more of the data structures as directed by one or more portions the software, according to particular needs.

In some embodiments, the described processing and memory elements (such as processors412and memory416) may be distributed across multiple devices such that the operations performed utilizing these elements may also be distributed across multiple devices. For example, software operated utilizing these elements may be run across multiple computers that contain these processing and memory elements. Other variations aside from the stated example are contemplated involving the use of distributed computing.

In addition or as an alternative, computer system400may provide particular functionality described or illustrated herein as a result of logic hardwired or otherwise embodied in a circuit, which may operate in place of or together with software to perform one or more steps of one or more processes described or illustrated herein. The present disclosure encompasses any suitable combination of hardware and software, according to particular needs.

Although the present disclosure describes or illustrates particular operations as occurring in a particular order, the present disclosure contemplates any suitable operations occurring in any suitable order. Moreover, the present disclosure contemplates any suitable operations being repeated one or more times in any suitable order. Although the present disclosure describes or illustrates particular operations as occurring in sequence, the present disclosure contemplates any suitable operations occurring at substantially the same time, where appropriate. Any suitable operation or sequence of operations described or illustrated herein may be interrupted, suspended, or otherwise controlled by another process, such as an operating system or kernel, where appropriate. The acts can operate in an operating system environment or as stand-alone routines occupying all or a substantial part of the system processing.

Although the present disclosure has been described with several embodiments, diverse changes, substitutions, variations, alterations, and modifications may be suggested to one skilled in the art, and it is intended that the disclosure encompass all such changes, substitutions, variations, alterations, and modifications as fall within the spirit and scope of the appended claims.