Abstract:
A method and system for scheduling jobs in a cluster of compute servers using dynamic job profiling improves the overall performance of the computing cluster. Specifically, the method and system monitors the performance of individual compute servers in a cluster, and assigns jobs to particular compute servers based on the most current performance information of the compute servers. As the job profiles of the compute servers change with every job processed, the decision as to which compute server is most appropriate for a particular job will dynamically change. Various scheduling algorithms can be used in connection with the method and system to assign jobs to specific compute servers based on the available information in the job profiles.

Description:
BACKGROUND OF THE INVENTION 
     Clusters of commodity computers are increasingly the platform of choice for running computationally intensive jobs in a variety of industries. Computations ranging from wind-tunnel simulations, gene and protein analysis, drug discovery, and many others are run on commodity computers with increasingly successful results. A typical cluster configuration as in  FIG. 1  may comprise a collection of compute servers  110   a . . . n , connected by a fast commodity network (typically 100 Mbps or 1 Gbps Ethernet), and a smaller number of machines acting as storage servers  120 . Users that want to use the system submit jobs through one or more gateway machines  130  that are responsible for providing an interface between the users  140  and the cluster network  100 , scheduling work on the cluster machines  110 ,  120  and returning the results of the jobs to the users. 
     The roles of machines  110 ,  120  in such a cluster need not be exclusive and membership of machines in the cluster may be transient or persistent. Most of the work done for such clusters to-date has focused on solving a number of important problems, such as, discovery of idle resources; management of job priorities; dealing with faults and the transient nature of compute servers; and automatic configuration of a smaller cluster out of a pool of resources based on an end-users description of their computational needs. 
     SUMMARY OF THE INVENTION 
     The present invention provides a method and system for scheduling jobs in a cluster of compute servers using dynamic job profiling. More specifically, the present invention monitors the performance of individual compute servers in a cluster, and assigns jobs based on the most current performance information of the compute servers. 
     Compute servers available to end users commonly have disparate characteristics. Disparity can take many forms including varying operating systems and processor architectures which would dictate what kind of executables a particular machine can run. Even in clusters where the operating systems and processor architecture is maintained uniform, other types of heterogeneity emerge as the cluster gets upgraded over time. For example, it is quite common for machines in a cluster to be upgraded piecemeal. Thus, a cluster can end up having compute resources with, among other things, varying processor speeds, disk I/O capabilities, memory sizes, file buffer cache sizes, floating point and integer arithmetic ratings. 
     An embodiment of the present invention uses existing technologies to monitor and store the job performance of compute servers in the network cluster, and uses the resulting job profile information to assign prospective or candidate jobs to a particular compute server. As the job profiles of the compute servers change with every job processed, the decision as to which compute server is most appropriate for a particular job will dynamically change. 
     In various embodiments of the present invention, the job scheduler uses different scheduling algorithms to assign jobs based on the available information in the job profiles. These algorithms include, but are not limited to, “greedy algorithms,” “competitive algorithms,” “affinity scheduling,” “reduction algorithms” or “genetic algorithms.” 
     Existing solutions focus on solving other important problems in the context of cluster scheduling, but do not try to learn anything about the properties of the jobs that are submitted to them. As a result they can make quite poor decisions on job placement and cause jobs to run significantly slower than necessary. These existing approaches to job scheduling fail to adequately address the unique opportunities presented by heterogeneous clusters and thus result in suboptimal use of resources in such settings. For example, a search of a DNA fragment against the human genome can take as little as 30 seconds or as much as 190 seconds depending on the amount of currently available memory that a machine has even if the difference between the total (overall) memory sizes of the two machines is no more than 10%. Furthermore, because certain data may already reside in a file buffer or cache, scheduling a job on a machine that has previously run a similar job can result in execution times 4-5 times faster than if jobs are randomly scheduled. 
     Based on these observations, in a cluster environment, the present invention constantly monitors jobs and takes information about jobs previously executed on the cluster to guide future scheduling decisions. The present invention complements the existing capabilities of cluster scheduling systems (i.e. resource discovery, fault tolerance, etc.), and allows cluster systems to become more intelligent and self directing. 
    
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS 
       The foregoing and other objects, features and advantages of the invention will be apparent from the following more particular description of preferred embodiments of the invention, as illustrated in the accompanying drawings in which like reference characters refer to the same parts throughout the different views. The drawings are not necessarily to scale, emphasis instead being placed upon illustrating the principles of the invention. 
         FIG. 1  is a cluster configuration in which the present invention may be employed; 
         FIG. 2  is a flowchart diagram of the present invention; and 
         FIG. 3  is a representative diagram of the job scheduler of one embodiment of the present invention. 
     
    
    
     DETAILED DESCRIPTION OF THE INVENTION 
     A description of preferred embodiments of the invention follows. 
       FIG. 2  shows a flow chart of an embodiment of the present invention. This embodiment operates in two components. The first component leverages existing technologies to introduce un-intrusive performance monitoring  240  on all compute servers (such as  110  of  FIG. 1 ) of a cluster and storing the gathered information in a job profile database  210 . Such monitoring tools already exist and can provide fairly detailed analysis of a job&#39;s computational profile with very little overhead (typically less than 5%). Furthermore they require no modifications to the executable being submitted and thus require no changes from an end-user&#39;s perspective. Examples of such monitoring tools are VTune® from Intel Corp or “Oprofile” for Linux systems. Other similar monitoring tools may be used at  240  in  FIG. 1 . 
     Additional information about a job&#39;s characteristics can be collected from a variety of system tools with little or no additional overhead as well. Some of the information collected in a job profile  210   a . . . n  may include, but is not limited to, the following: files being accessed and amount of I/O per file (in bytes and number of I/O operations or calls), floating point to integer instruction ratio, number of instructions executed, memory working set size, cache working set size, number of bytes collected from disk, and network I/O accesses. 
     Preferably, for each job the executable&#39;s name, the arguments passed to it and the execution profile (i.e. machine type used, cache type, total run time, etc.) as it has been collected by performance monitoring tools is saved in the job profiles  210   a . . . n . In addition, job profile database  210  stores/records information regarding the characteristic of each machine in the cluster (using a small set of benchmark programs) with respect to its floating point and integer capabilities, its file buffer cache size and type (hot or cold cache, etc.), its memory size, type of disk memory, and the capabilities of its I/O and networking subsystems. 
     As such, the job profiles  210   a . . . n  and entries in database  210  provide groupings or categories of jobs based on previously executed jobs. Per job category, there is at least one job profile and a preferred machine according to the recorded statistics and performance information of the previously executed jobs. Restated, the system of the present invention enables each compute server machine  110  to be associated with the job categories and hence the profiles of jobs for which the computer server has most efficiently processed (historically and/or recently). Thus given a candidate job to be scheduled for execution, the job category and/or closest matching job profile of the various job categories represented in database  210  have determined the corresponding preferred machine for executing such jobs (i.e. jobs that fit or match the job profile or that job category of the given job). 
     The second component  220  of the present invention uses the job profile information and more specifically the job category information held in database  210  to attempt to improve the overall throughput of the cluster by making intelligent job placement decisions. While an exact online solution to the scheduling problem is impossible, reasonable heuristics can be found that can significantly improve throughput. As an example, jobs  230  that access the same large database files with different search patterns can run 4-5 times faster on a machine that has pre-warmed the database file, and therefore it is beneficial to wait for such a machine to become available unless the expected wait time exceeds the difference in expected runtime savings. Depending on the nature of the executable names and arguments of a given candidate job, and also the nature of the capabilities of the cluster network servers (as indicated in the saved information in database  210 ), a different algorithm can be used by job scheduling module  220  in connection with the dynamically changing job profiles  210   a . . . n  to optimize performance. 
       FIG. 3  provides a representative diagram of such a job scheduler of an embodiment of present invention. A candidate job  301  is provided with the names of a respective executable, arguments, and argument values. When the job  301  gets submitted to the cluster gateway  310 , the gateway uses the executable&#39;s name as a key into the local database of the invention system  300  in order to look up information  320  for this particular executable. Known or common database query  320  techniques are employed. In addition to determining the closest matching job profile and in turn job category and corresponding preferred compute server, the information returned  330  can be a list of arguments for this executable that have been observed to have a material impact on job execution time in the past. The current values of those arguments are compared with previous values for which the system  300  has accumulated statistics. Furthermore, information can be queried  320  and retrieved  330  on which machines a job with the same executable and similar arguments was last run. 
     Based on the foregoing query results and retrieval information, an algorithm  340  determines to which compute server to send the job  301 . On output  350 , the scheduling module  220  places the incoming or candidate job  301  on a machine in the cluster network  100  that either has run a similar job recently or if no such machine exists, on a machine whose resource specification best matches the expected behavior of the incoming job  301 . If a suitable machine exists but is currently busy it may still be preferable to wait for that machine to finish its work and schedule the new incoming job  301  there, rather than scheduling it on an already idle machine. 
     Algorithm  340  may apply affinity scheduling for assigning database transaction or parallel processing. Pseudocode for the algorithm  340  of one embodiment is as follows for purposes of illustration and not limitation.
         Given machine M 1 , Job B (the job type and category last run on M 1 ) and Input Job A (to be scheduled):
 
 t   wait =½[( t   A,M1,Ccold )−( t   A,M1,Chot )+( t   B,M1,Ccold )−( t   B,M1,Chot )]
   Where (t A,M1,Ccold ) is the time Job A takes on M 1  when the File Buffer Cache is cold.   Where (t A,M1,Chot ) is the time Job A takes on M 1  when the File Buffer Cache is hot.   Where (t B,M1,Ccold ) is the time Job B takes on M 1  when the File Buffer Cache is cold.   Where (t B,M1,Chot ) is the time Job B takes on M 1  when the File Buffer Cache is hot.   Job A waits t wait  to run on M 1 , otherwise it will schedule Job A to run on another machine.       

     It is understood that a variety of algorithms  340  of various complexity may be employed. For example, other constraints may be added to the foregoing algorithm  340 . In addition, rule based techniques, greedy algorithms, competitive algorithms, M Queue algorithms, and the like may be utilized singularly or in combination. 
     After the selected machine has processed the candidate job  301 , the system  300  updates the job profiles in the job profile database  210  as described above in  FIG. 2 . 
     The behavior of the system  300  of the present invention is not static, but varies over time. For example, some jobs may change the state of databases that they are working on and thus the same job with the same argument can have a different runtime the next time it is run. As long as the changes are gradual, the monitoring sub-system/module  240  will detect the differences  360  and update  370  the job profiles  210   a . . . n  which ultimately adjusts the scheduling policies (at  340  and  220 ) appropriately. Furthermore to ensure that the scheduling system  300  does not get stuck in suboptimal decisions, small amounts of randomness may be introduced in the scheduling  220  so that new profiles can be collected about jobs even after the jobs have been studied for a long time. The amount of randomness in the scheduling can depend on the history of a job (the more history the less randomness needed), and the relative values of more recent runs versus older runs (i.e. if a new non-random run produces markedly worse runtimes then this is a sign that random scheduling and profile collection are currently needed). 
     This embodiment of the present invention increases the throughput capacity of a fixed collection of machines  110  given a particular workload. Unlike previous approaches which tend to focus on resource utilization rather than throughput maximization, this embodiment of the present invention, makes the best use of compute resources while focusing on average job turn around time. This approach results in higher productivity from the end-user&#39;s perspective, and better return on investment for the IT manager that controls the decisions on computing purchases. Furthermore, a profile drive scheduler can provide a differentiator in what is otherwise a commodity field and make the difference between one vendor&#39;s cluster solution versus another. 
     Embodiments of the present invention involve modifications to the cluster scheduling system  300  that are invisible from an end user&#39;s perspective and thus, can be adopted with little or no change in other operational aspects of the system. Furthermore, even in the absence of sufficient information for intelligent decisions, the performance degradation of the system of the present invention over a more traditional schedule is a very modest 3-5% due to our monitoring and profile collection activities. On the other hand, the upside in successful use cases is 4- or 5-fold increase in throughput. Given these tradeoffs, the performance penalty paid in the failure cases is a small price to pay in order to realize the potential gain. 
     While this invention has been particularly shown and described with references to preferred embodiments thereof, it will be understood by those skilled in the art that various changes in form and details may be made therein without departing from the scope of the invention encompassed by the appended claims.