Application server scalability through runtime restrictions enforcement in a distributed application execution system

In an application execution system having a plurality of application servers, each application server includes non-volatile storage for storing a plurality of applications, volatile storage and computational resources for executing applications in response to requests received by the application execution system and for returning results to the requestors. In addition, each application server includes instructions for conditionally terminating execution of the respective application prior to returning the result if the respective application violates any of a plurality of execution restrictions. The execution restrictions in aggregate maintain scalability of the application execution system with respect to at least one or more of the following limits: number of applications that can be executed and number of requests that can be processed per period of time.

RELATED APPLICATIONS

This application is related to U.S. patent application Ser. No. 12/060,798, filed Apr. 1, 2008, entitled “Efficient Hosting in a Distributed Application Execution System” which is incorporated by reference herein in its entirety.

TECHNICAL FIELD

The disclosed embodiments relate generally to methods and systems, sometimes called application servers, for hosting and executing large numbers of heterogeneous applications.

BACKGROUND

In general, increases in an application's popularity could present a variety of scalability problems that negatively impact a user's experience. For example, users could experience slower response times, slower page loading, and increased time outs on page requests. These scalability problems are typically alleviated by allocating additional capacity to the application such as more storage, more memory, more CPUs, and more machines in general.

Allocating or installing more computing capacity may be a reasonable solution when increases in an application's popularity are experienced over a prolonged period of time, or when usage of the application is predictable. Similarly, when an application experiences a decrease in usage, removing computing capacity previously allocated to the application may be a reasonable solution, especially when the decrease is experienced over a prolonged period of time, or when the decrease is predictable. However, the popularity of an application is often unpredictable, due to a variety of factors (e.g., time of day, current events, advertising, trends), and fluctuates to a large extent, which creates load spikes and dips in the application execution or hosting system.

Predefined allocations of computing resources are inefficient solutions for handling temporary load spikes and dips. Increasing or installing more computing resources to handle a load spike is inefficient, since the additional pre-allocated resources go unused when the spike disappears (e.g., when the spike in demand subsides, or the application's popularity dips). Similarly, decreasing computing resources allocated to an application when its popularity declines is also inefficient, since future usage spikes will require the re-allocation of previously removed resources back to the application.

To complicate matters further, application systems may host a large number of heterogeneous applications, each with its own set of fluctuating resource requirements. Pre-allocation of resources, for the reasons discussed above, is often an inefficient solution for ensuring consistent positive user experiences among heterogeneous applications hosted on an application system.

SUMMARY

In an application execution system having a plurality of application servers, each application server includes non-volatile storage for storing a plurality of applications, volatile storage and computational resources for executing applications in response to requests received by the application execution system. Each application server also includes instructions for storing in non-volatile storage applications distributed to the application server, for loading a plurality of respective applications into volatile storage, for executing a respective application in response to a respective received request from a client, and for returning a result to the request. In addition, each application server includes instructions for conditionally terminating execution of the respective application prior to returning the result if the respective application violates any of a plurality of execution restrictions. The execution restrictions include two of more of the following: a response time limit; an average response time limit over multiple executions of the respective application; a volatile memory usage limit that limits how much volatile memory can be used by each execution of the respective application; a restriction on using non-volatile storage of the application server; and a requirement that any state information to be retained after returning the result is either sent to the client for storage, or retained in a datastore accessible by all the application servers. The execution restrictions in aggregate maintain scalability of the application execution system with respect to at least one or more of the following limits: number of applications that can be executed and number of requests that can be processed per period of time.

In another aspect of the invention, a method of operating an application server in a system having a plurality of application servers is disclosed. The method comprises storing in non-volatile storage applications distributed to the application server, loading a plurality of respective applications into volatile storage, executing a respective application in response to a respective received request from a client, and returning a result to the request. In addition, the method includes conditionally terminating execution of the respective application prior to returning the result if the respective application violates any of a plurality of execution restrictions. The execution restrictions include two of more of the following: a response time limit; an average response time limit over multiple executions of the respective application; a volatile memory usage limit that limits how much volatile memory can be used by each execution of the respective application; a restriction on using non-volatile storage of the application server; and a requirement that any state information to be retained after returning the result is either sent to the client for storage, or retained in a datastore accessible by all the application servers. The execution restrictions in aggregate maintain scalability of the application execution system with respect to at least one or more of the following limits: number of applications that can be executed and number of requests that can be processed per period of time.

DESCRIPTION OF EMBODIMENTS

FIG. 1is a block diagram of a distributed computing system100including an application execution system130connected to a plurality of clients102(e.g.,102-1. . .102-n) through a network110such as the Internet, other wide area networks, local area networks, metropolitan area networks, wireless networks, or any combination of such networks. In some embodiments, a respective client102contains one or more client applications104(e.g.,104-1. . .104-n), such as a web browser, for submitting application execution requests to the application execution system130. The client102(sometimes called the “client device” or “client computer”) may be any computer or similar device through which a user of the client102can submit requests to and receive results or services from the application execution system130. Examples include, without limitation, desktop computers, notebook computers, tablet computers, mobile devices such as mobile phones, personal digital assistants, set-top boxes, or any combination of the above.

In some embodiments, the application execution system130include a front-end server140, which receives application execution requests from clients102and returns results to the requesting clients. The front-end server140receives, from the application master150, an application distribution map which may include resource usage information that can be used to route requests received from client(s)102. More information regarding the application distribution map is provided in U.S. patent application Ser. No. 12/060,798, filed Apr. 1, 2008, entitled “Efficient Hosting in a Distributed Application Execution System” which is incorporated by reference herein in its entirety.

The application execution system130include a plurality of application servers160(e.g.,160-1through160-n). As described in more detail below with reference to FIG.5, each of the application servers160includes non-volatile storage for storing a plurality of applications in a local library162, volatile storage, and computational resources for executing applications in response to requests received by the application execution system130. The application servers160may include a runtime manager164, which as further discussed below with reference toFIGS. 2 and 3C, enforces resource limits and restrictions for various resources consumed by an application during execution. In some embodiments, the application execution system130includes a quota system180which, in conjunction with the runtime manager164, implements a plurality of application restriction limits as further discussed below.

The application execution system130also includes an application master150that distributes applications, from a main library152having a plurality of applications, among the application servers160. In the embodiment shown inFIG. 1, the main library152is stored in the application master150. Alternately, the main library152may be stored remotely from the application master, such as in a datastore170. In some embodiments, each application of the plurality of applications in the main library152is a web application that is responsive to HTTP requests. However, the present invention can also be used in non-web based environments, in which case the applications need not be web-based applications.

In some embodiments, the application execution system130also includes a datastore170accessible to at least the application master150and the front-end server140, for sharing information about the location of applications among the application servers160and resource usage or loading information with respect to the application servers160.

Optionally, the distributed system100includes additional resources106, which may be located either internally or externally to the system100, for use when executing applications in the application execution system130. For example, an application executed by the application execution system130may access information in one or more of the additional resources106in order to process a request received from a respective client102. These additional resources may include one or more of: other applications and data provided by web services (e.g., web feed data from sources such as blog entries, headlines, podcasts, etc.).

FIG. 2is a flow diagram illustrating a method200performed by an application server according to some embodiments. The applications distributed to the application server by the application master150are stored in the application server's non-volatile memory (202). Assuming that there is sufficient volatile memory, in response to an application request (204) from the front-end140, the application server160loads the requested application into volatile storage, executes the requested application, and returns a result to the request (212). In some embodiments, the application servers prevent or terminate execution of the application if the application violates any of the execution restrictions further discussed below.

The application server may remove one or more applications from volatile memory before performing operation212when there is insufficient volatile memory to process the request. Alternatively, the application server may automatically remove applications from volatile memory when the amount of available volatile storage is below a predefined threshold.

In some embodiments, after returning the result to the request, the application server can either remove the respective application from volatile storage (218), or retain the respective application in volatile storage (220) for responding to future requests, by determining whether predefined criteria has been met (214). In some embodiments, the predefined criteria (216) used by the application server include usage level criteria, which may include the number of requests for the application processed per period of time. The predefined criteria may also include caching criteria, which may include the number of respective application instances in volatile storage available for handling new requests, and the number of respective application instances handling active requests and therefore not available for handling new requests. Cached application instances are discussed in more detail below with reference toFIG. 3A. The predefined criteria may also include error criteria, which may be based on the number of errors encountered during execution of the application, and the type of errors encountered. For example, the application server may remove the respective application from volatile storage if severe errors are encountered during N (e.g., 5, 10 or 20) consecutive executions of the application.

For applications that fail to meet the predefined criteria (214—No), the application server removes the respective applications from volatile storage upon returning the result to the request (218). In some embodiments, the application server may remove the respective applications from volatile storage according to a predefined order for removal. For example, the application server may remove the least recently used application.

In some embodiments, when determining which application instance to remove from volatile storage, the application servers may take into account the service quality levels of the applications for which instances are stored in volatile memory. The service quality level of each application may be based on the level of service requested, or paid for. Various forms of preferences (for retention of application instances in volatile memory) may be given to applications with high service quality levels, compared to applications with lower service quality levels. For example, lower service quality level application instances may be evicted before higher service quality level application instances whenever a predefined condition is true. The predefined condition may relate to numbers or ratios of lower and higher service quality level application instances loaded in volatile memory. Alternately, scores may be computed to determine which application instances to unload from volatile memory, and computation of the scores may take into account the service quality levels of the applications.

For applications that meet the predefined criteria (214-Yes), the application server retains the respective applications in volatile storage for responding to future requests for the application (220). In some embodiments, the application server retains more than one application instance of the respective application in volatile storage in accordance with predefined caching criteria. In some embodiments, the application server limits the number of application instances in volatile storage. For example, the application server may limit the total number of application instances in volatile memory to ensure that there is sufficient volatile memory for other processing tasks. Alternatively, the application server may limit the number of instances of a respective application to ensure that other requested applications have access to sufficient volatile memory to service their requests.

In some embodiments, the application server may terminate (222) execution of the respective application prior to returning the result if the respective application violates any of a plurality of execution restrictions. These execution restrictions may include application resource limits for limiting an application's consumption of system resources during runtime (e.g., a response time limit, an average response time limit over multiple executions of an application, a volatile memory usage limit that limits the amount of volatile memory used by each execution of an application). In some embodiments, the same application resource limits are applied to all applications (e.g., the volatile memory usage limit is the same for all applications). For example, the application execution system may terminate execution of an application if the application's run time or execution time exceeds the response time limit. Alternatively, the application resource limits may vary between applications. Optionally, in embodiments where the average response time for an application is frequently or periodically updated, the application execution system may terminate execution of the application if the application's average response time exceeds the average response time limit (e.g., 0.5 seconds). For example, multiple instances (e.g., all instances being executed) of the same application may all be terminated if, for some reason, the average response time of the application extends beyond the applicable resource limit. The application execution system may also restrict the number of requests processed per period of time for an application, the number of CPU cycles per period of time, the stack size associated with the execution of an application, and other resource usage data as further discussed with reference toFIG. 4(Resource Usage Data434).

The execution restrictions may also prohibit certain actions performed by applications using one or more predefined application programming interfaces (APIs). These execution restrictions may restrict the use of non-volatile storage of the application server by an application during execution by preventing the application code from reading or accessing data stored in non-volatile storage. In addition the execution restrictions may govern the storage of state information by requiring state information (to be retained after returning the result) to either be sent to the client for storage, or retained in a datastore accessible by all application servers in the application execution system. Other execution restrictions may include restrictions on: opening network connections; creating new processes by an application, including the creation of threads by the application; making system calls by an application; and other system function calls. In some embodiments, to maintain scalability of the application execution system with respect to number of applications that can be executed and number of requests that can be processed per period of time, the execution restrictions include a restriction requiring all applications to execute within a secure execution environment. In these embodiments, external access to memory locations within the application servers is prohibited. The application execution system may also monitor and enforce restrictions on the frequency of certain actions performed by the applications that impact system performance by monitoring the frequency of calls (made by an application) to APIs associated with these restricted actions (e.g., number of times per period of time that the application calls the API(s) for accessing a datastore, number of times per period of time that the application calls the API(s) for accessing an external web service).

As previously discussed with reference toFIG. 1, in some embodiments, the application execution system includes a quota system180. The monitoring and enforcement of certain execution restrictions described above may involve the quota system180, which maintains aggregate quota information for each application. When an application needs to consume a system resource, the application execution system may consult the quota system to ensure that the application has not exceeded quota or threshold limits. For example, the system may include a restriction on the number of times per day the application can access the datastore (e.g., 2,000,000 calls per day) using the associated API (e.g., the Datastore Interface572FIG. 5). Before the application can call the Datastore Interface572, the system checks the number of calls to the Datastore Interface572(made by a respective application on all application servers) to ensure that the application is still within its quota limit (e.g., 2,000,000 calls per day).

FIG. 3Ais a conceptual diagram of a process for managing application instances in volatile memory by the application server in accordance with some embodiments of the invention. The application servers actively manage application instances in volatile memory through a live process cache350. The application server first initiates a generic application instance330, sometimes called a “generic instance.” The generic instance is “cloned” multiple times (332) resulting in a “pool of clones” (334) of the generic instance. When a request for an application is received, the application server removes (336) a generic instance clone from the “pool of clones” and loads (338) the code for the requested application into the generic instance clone to produce an application instance342. In some embodiments, when a clone is removed from the “pool of clones,” a request is issued (335) to create a new generic instance clone to replace the one selected. The code for the requested application may be associated with any of a variety of programming languages including but not limited to: Java, Perl, Python, or C. In some embodiments, the application is executed using an interpreter, while in other embodiments the application code is compiled and linked (as needed) with any other code (e.g., library procedures) to produce code that is ready for execution by the application server, in the context of the application server's operating system.

InFIG. 3A, the application server loads (338) the code for “App1” into a generic instance in response to a request received for “App1.” The “App1” instance may be referred to as “live” or “alive” since it contains code that is readily executable by the application server. The App1instance is loaded into a cache (e.g., “Live Process Cache”)350. The cache may contain other instances of App1as well as instances of other requested applications (e.g., App2. . . AppN). These application instances in the Live Process Cache350are either “busy” or “available,” although other transitional states could be used in addition in other embodiments. A “busy” application instance is one that is processing a current request, and therefore cannot process another incoming request until it finishes processing the current request. The “available” instances are ready to process any received request for the respective application.

FIGS. 3B-3Cis a flow diagram illustrating an example of a runtime process for handling a request for an application at the application server in accordance with some embodiments. For the purposes of this example, it may be assumed that the application server has sufficient resources available to handle an incoming request to execute an application. Upon receiving (304) a request from the front-end for “App1,” the application server queries the cache350(FIG. 3A) for an available “App1” instance for handling the request. If the cache returns an available “App1” instance (362-Yes), the request is forwarded (364) to the returned instance for processing. If the cache does not return an available “App1” instance (362—No), the application server determines (366) if the “pool of clones”334has at least one generic instance. Other application instances (e.g., the least recently used instance) are removed from the cache if necessary (e.g., to make room for new application instances). If there are no generic instances available (366—No), the process waits (368) for a generic instance clone to be available before performing operations (372) and (374). If there are generic instances available (366-Yes), a generic instance clone is selected (372) from the “pool of clones.” The App1code is loaded (372) onto the selected generic instance, and the App1instance is loaded (374) into the live process cache350. The App1instance is then executed (382) in volatile memory. During execution of App1, a plurality of execution restrictions are monitored (384), such that the application server may terminate (388) execution of App1if one or more execution restrictions are violated (386-Y). In addition, upon termination of the application, the application server may return an error message to the requesting client. In some embodiments, application server also removes (392) the terminated App1instance from volatile storage (e.g., Live Process Cache350). If no execution restrictions are violated (386-N) during execution of App1, App1completes processing the request and returns (390) a result.

FIG. 4is a block diagram of a table or other data structure (live process ca cache data)562for storing information relating to application instances stored in volatile memory (e.g., Live Process Cache350,FIG. 3A) for processing application requests in accordance with some embodiments of the invention. As shown inFIG. 4, live process cache data562includes the following items of information, or a subset or superset thereof:cache size information408, which may be represented (for example) in terms of memory capacity (e.g., a number of gigabytes), or a maximum number of application instances that can be stored in the live process cache350;the number of generic instances410currently available for allocation; andinformation412about each of the applications, or application instances, currently in the live process cache350. For example, information412may indicate for each application having at least one instance in the live process cache350, the identity424of the application, the number426of “busy” instances of the application, the number428of “available” instances of the application, last use information430(e.g., the date/time of last use of the application), and data for execution restrictions432or other information for enabling implementation of a cache eviction policy. Examples of execution restrictions data432including resource usage data434, response time data436, and violations data438are further described below.

Resource usage data434, which allows the system to monitor the amount of system resources consumed by an application, may include any combination of the following (per period of time): number of CPU megacycles, number of HTTP requests, amount of bandwidth consumed by incoming requests for the application (“bandwidth in”), and amount of bandwidth consumed by transmission of outgoing results to requests for the application (“bandwidth out”). In addition, resource usage data434may also monitor the frequency of certain activities such as accessing a datastore, accessing internal or external application resources106(e.g., web services, web feeds, news feeds), and sending email from the application. These monitored activities are further discussed above with reference toFIG. 2.

Response time data436includes data indicating the amount of time it takes the application to issue a response to serve a request. Violations data438may include data indicating the frequency of restriction violations associated with the application (e.g., number of response time violations over a period of time, type of violated restriction(s), aggregate number of violations). In some embodiments, violations data438is updated during application execution. For example, during execution of an application, the application server (executing the application) may monitor and update violations data438to ensure compliance with application restrictions.

FIG. 5is a block diagram of an application server500, which may be implemented using one or more servers, in accordance with some embodiments of the invention. For ease of explanation, the application server500is herein described as implemented using a single server or other computer. The application server500generally includes one or more processing units (CPU's)502, one or more network or other communications interfaces506, memory510, and one or more communication buses504for interconnecting these components. The communication buses504may include circuitry (sometimes called a chipset) that interconnects and controls communications between system components. Memory510may include high speed random access memory and may also include non-volatile memory, such as one or more magnetic disk storage devices. Memory510may include mass storage (e.g., datastore170,FIG. 1) that is remotely located from the central processing unit(s)502. Memory510, or alternately the non-volatile memory device(s) within Memory510, comprises a computer readable storage medium. In some embodiments, Memory510stores the following programs, modules and data structures, or a subset thereof:an operating system512that includes procedures for handling various basic system services and for performing hardware dependent tasks;a network communication module514that is used for connecting the application server500to other computers via the one or more communication network interfaces506(wired or wireless) and one or more communication networks (e.g., communication networks110), such as the Internet, other wide area networks, local area networks, metropolitan area networks, and the like;an application server module516for processing application requests. In some embodiments, the application server module516includes an application master interface module518for interfacing with the application master150(FIG. 1), a front-end interface module for interfacing with the front-end140, a quota system interface module529for interfacing with the quota system180, and a live process cache monitor and control module532for managing application instances in the live process cache350(FIG. 3A). The application server module516may also store a security module544for executing untrusted processes (Run Untrusted Process Module)548and trusted processes (Run Trusted Process Module)546. Some of the procedures included in the application server module516are further described below.a local application library162, for storing the applications distributed to the application server by the application master150;application instances in volatile memory550(e.g., in a live process cache350,FIG. 3A) for servicing application requests. In some embodiments, there is at least one application instance for an application in volatile memory;data552includes data used primarily by the application master150(e.g., local application library data554, server resource data556, application resource data558). More information regarding data used by the application master is provided in U.S. patent application Ser. No. 12/060,798, filed, Apr. 1, 2008, entitled “Efficient Hosting in a Distributed Application Execution System” which is incorporated by reference herein in its entirety. When needed, data552includes user data560, which may include data received from a requesting user (e.g., user name, passwords, user preferences, profiling information) and/or data produced or retrieved for the requesting user by the application server. In addition, data552may include live process cache data562, described with reference toFIG. 4;an optional Cache Memory580, for temporarily storing data and making the data for a respective application accessible to all instances of that application.

In some embodiments, the application master interface module518includes procedures for adding or removing applications from the non-volatile storage of the application server. The application master interface module518may also include procedures for sending usage information on application resources and server resources to the application master. In some embodiments, the front end interface module526includes procedures for handling application requests528forwarded from the front end140. More information regarding the application master and the front-end is provided in U.S. patent application Ser. No. 12/060,798, filed, Apr. 1, 2008, entitled “Efficient Hosting in a Distributed Application Execution System” which is incorporated by reference herein in its entirety.

Procedure(s) (Live Process Cache Monitor & Control)532for managing application instances in the live process cache350may include procedures (Load and Remove Application Instance Module)534for loading and removing application instances into the live process cache in accordance with application usage and available volatile memory as previously discussed with reference toFIG. 2, procedures (Execute Application Instance Module)536for executing application instances when processing application requests. The Execute Application Instance Module536may also include procedures (Resource Limit Enforcement)538for limiting resource consumption of a particular application. For example, an application that consumes more resources than a limit or threshold may be terminated (Terminate App Instance)540, as previously described with reference toFIG. 2. The resource limit may be a predefined amount or the threshold may vary depending on factors such as the number of requests for the application. For example, applications that receive higher numbers of requests may have a higher threshold before the application instance is terminated. Alternatively, the threshold may also depend on the amount of processing resources (e.g., one or more of: CPU time, “wall clock” time (i.e., total elapsed real time), memory, communication bandwidth, and number of system function calls made) consumed by the application. The threshold(s) may be applied per execution of an application, or to a running average of resources used over multiple executions of the application. An application instance that consumes resources above a corresponding threshold may be terminated.

The procedures (Live Process Cache Monitor & Control)532may also include application programming interfaces (APIs)570that enable and restrict activities that applications may engage in during execution. The APIs570may include any combination of the following: a Datastore Interface572for interfacing with a datastore (e.g., the datastore170); a Cache Memory Interface576for interfacing with shared Cache Memory580; and an External Resources Interface574for interfacing with external resources (e.g., other applications, other websites, web services). In embodiments having the Cache Memory580and the Cache Memory Interface576, an application instance can store data (e.g., data copied from a datastore) for high speed access by all instances of the same application. The application must not rely upon retention of data by the Cache Memory580; for example, the data stored in the Cache Memory580for a respective application may be deleted whenever there are no longer any active instances of the application. The Datastore and Cache Memory mechanisms for sharing information between instances of the same application are accessibly only through the APIs provided by the application execution system, which facilitates enforcement of application restrictions associated with both durable and temporary storage of data that can be shared between application instances.

The live process cache monitor & control module532may also include procedures for monitoring the status of application instances (Monitor Application Instance Status Module)542. For example, the status of the application instances may be “busy,” “available,” or any transitional state in between (seeFIG. 3A).

FIG. 5shows an application server system and is intended more as a functional description of the various features which may be present in a set of servers than as a structural schematic of the embodiments described herein. In practice, and as recognized by those of ordinary skill in the art, items shown separately could be combined and some items could be separated. For example, some items shown separately inFIG. 5could be implemented on single servers and single items could be implemented by one or more servers. The actual number of servers used to implement each such subsystem and how features are allocated among them will vary from one implementation to another, and may depend in part on the amount of data traffic that the system must handle during peak usage periods as well as during average usage periods.