Task allocation multiple nodes in a distributed computing system

Field

The present disclosure relates to distributed computing, and more specifically relates to allocation of tasks amongst nodes in a distributed computing environment.

Background

In the field of distributed computing, individual computers are typically linked together to form a computing cluster for the purpose of solving computationally-extensive problems. A computing cluster typically consists of a front end node and multiple compute nodes. Under this arrangement, the front end node typically allocates a problem to the compute nodes by dividing the problem into multiple tasks and distributing those tasks amongst the compute nodes for processing.

According to one technique for task allocation, the front end node divides a problem into equally-sized tasks and distributes the tasks to the compute nodes for processing. Upon distribution of the tasks, the front end node waits for the compute nodes to complete their assigned tasks, and to transmit the results back to the front end node for further processing.

Summary

One difficulty with such implementations typically arises when a computing cluster consists of compute nodes that vary in terms of hardware and software resources. In such clusters, all compute nodes are typically assigned a task of the same size, irrespective of the hardware and software capabilities of the individual compute nodes. Naturally, resource-rich compute nodes complete their tasks faster than other compute nodes. As a consequence, the front end node must wait for the slower compute nodes to finish their tasks before the next set of tasks can be assigned to the compute nodes. Meanwhile, the faster compute nodes, which have already completed their tasks, remain idle, thus reducing the overall utilization of the computing cluster.

One technique for overcoming this issue is to assign new tasks to the faster compute nodes immediately upon completion of their previously-assigned tasks, thereby preventing the faster nodes from remaining idle. However, this solution is often inadequate because completion of the overall problem is still constrained by the slowest node, which might have been assigned a task that is too large for its capabilities.

The foregoing situation is addressed herein by sizing tasks based on a dynamically adjusted job load metric, where the job load metric is dynamically adjusted responsive to a time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task.

Thus, in an example embodiment described herein, work is distributed amongst a plurality of nodes in a distributed computing environment. A first plurality of tasks is extracted from the work, where the number of tasks in the first plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the first plurality of tasks are sized based on a job load metric of each of the nodes. The first plurality of tasks is distributed to at least some of the nodes for processing. A determination is made whether a time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task exceeds a predefined threshold. In response to a determination that the time difference exceeds the predefined threshold, the job load metric of at least some of the nodes is adjusted.

Thereafter, a second plurality of tasks is extracted from the work, where the number of tasks in the second plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the second plurality of tasks are sized based on the adjusted job load metric of each of the nodes. The second plurality of tasks is distributed to at least some of the nodes for processing.

By sizing tasks based on a dynamically adjusted job load metric, it is ordinarily possible to size the tasks such that, after several iterations, all tasks tend to complete at around the same time, within the predefined threshold. As a consequence, there is a reduction in the amount of time a front end node must wait between responses from the compute nodes, thereby resulting in a decrease in compute node idle times and an increase in computing cluster utilization.

In another example embodiment, a qualification task is transmitted for execution by each node, where the qualification task measures capabilities of core elements in a node, where an influence factor is assigned to each core element, and where the influence factor is used to adjust the weight of the core element on a capability measurement of the node. In particular, the results from the executed qualification task from each node are received and the job load metric for each node is calculated by using the results from the executed qualification task.

In another example embodiment, the distribution of work amongst a plurality of nodes in a distributed computing environment comprises a repetition of the steps of determining whether a time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task exceeds a predefined threshold, adjusting the job load metric of at least some of the nodes responsive to a determination that the time difference exceeds the predefined threshold, extracting a second plurality of tasks from the work, where the number of tasks in the second plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the second plurality of tasks are sized based on the adjusted job load metric of each of the nodes, and distributing the second plurality of tasks to at least some of the nodes for processing.

In still another example embodiment, the work available for distribution is dynamic in nature such that the amount of work available for distribution is received on an ongoing basis, and where the work available for distribution is distributed and processed in the order received.

In yet another example embodiment, responsive to a determination that the time difference does not exceed the predefined threshold, a second plurality of tasks is extracted from the work, where the number of tasks in the second plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the second plurality of tasks are based on the unadjusted job load metric of each of the nodes and the second plurality of tasks is distributed to some of the nodes for processing.

In a further example embodiment, adjusting the job load metric comprises dynamically adjusting capacity utilizations for at least some of the core elements in a node during runtime, and adjusting the job load metric using the dynamically adjusted capacity utilizations of the core elements in the node.

In another example embodiment, dynamically adjusting capacity utilizations for at least some of the core elements in a node during runtime comprises reducing the capacity utilizations for a plurality of core elements in the worst performing nodes, and increasing the capacity utilizations for a plurality of core elements in the remaining nodes, wherein adjustments to the capacity utilizations for the plurality of core elements are made in descending order of the influence factor assigned to the core elements.

In yet another example embodiment, dynamically adjusting capacity utilizations for at least some of the core elements in a node during runtime comprises reducing the capacity utilization for a core element assigned the highest influence factor in the node that is last to complete its task, by a predetermined amount, and increasing the capacity utilization for the core element assigned the highest influence factor in the remaining nodes, wherein the amount increased in each node is proportionate to the existing capacity utilizations for the core element assigned the highest influence factor in the remaining nodes, and wherein the cumulative increase amongst the remaining nodes is equal to the amount that was reduced in the node that is last to complete its task.

This brief summary has been provided so that the nature of this disclosure may be understood quickly. A more complete understanding can be obtained by reference to the following detailed description and to the attached drawings.

Brief description of the drawings

FIG. 1 is a representative view of a computing cluster relevant to one example embodiment.

FIG. 2 is a detailed block diagram depicting the internal architecture of an example front end node.

FIG. 3 is a detailed block diagram depicting the internal architecture of an example compute node.

FIG. 4 illustrates example architecture for a work distribution module in a front end node and example architecture for a work processing module in a compute node, and interaction therebetween.

FIG. 5 is a flowchart illustrating an example process for explaining task allocation.

FIG. 6 is a flowchart illustrating an example process for explaining task execution.

FIG. 7 is a view for explaining the allocation of tasks for successive frames of image data in moving image data.

Detailed description

FIG. 1 is a representative view of a computing cluster relevant to one example embodiment.

As shown in FIG. 1, front end node 200 is connected to compute nodes 300a to 300d via network 1000. The front end node 200 extracts and distributes tasks to the compute nodes 300a to 300d. As depicted diagrammatically in FIG. 1, front end node 200 is provided with multiple time-sequential frames of image data in a moving image 100, such as frames 101 to 105, for which extraction and distribution to compute nodes 300a to 300d is effected. The frames 101 to 105 are received in a work queue 222, where they are distributed and processed in the order received.

The present disclosure contemplates various numbers of compute nodes. Accordingly, more or fewer compute nodes may be implemented in the computing cluster, and the number of participating compute nodes might change during work.

In FIG. 1, front node 200 is depicted as receiving work, such as a moving image 100, via the Internet (or intranet) for processing. The front end node 200 divides each frame 101 to 105 into multiple tasks for distribution and processing by multiple compute nodes 300a to 300d, such that each frame 101 to 105 is processed by more than one compute node. Thus, as depicted in FIG. 1, with respect to frame 101, the front end node 200 divides a plurality of tasks 101a to 101d, and distributes the divided tasks 101a to 101d to the compute nodes 300a to 300d for processing. As also depicted in FIG. 1, compute node 300a is assigned task 101a, compute node 300b is assigned task 101b, compute node 300c is assigned task 101c, and compute node 300d is assigned task 101d. The size of each divided task corresponds to a job load metric of the compute nodes that is assigned to execute that task, as discussed more fully below in connection with FIG. 5.

The compute nodes 300a to 300d transmit the results of processing to the front end node 200 upon completion of their assigned tasks. The front end node 200 receives responses from the compute nodes 300a to 300d, including results of processing, where the results may be transmitted to a server for further processing. According to the embodiment shown in FIG. 1, the front end node 200 transmits the results of processing to server 106, for display by the high-resolution output device 107.

The particular implementation, scale, and hardware of network 1000 may vary according to different embodiments. Thus, for example, network 1000 could be the Internet, a Local Area Network (LAN), Wide Area Network (WAN), Metropolitan Area Network (MAN), Personal Area Network (PAN), or combinations thereof. Network 1000 can be wired or wireless, and can be implemented, for example, as an Ethernet, Infiniband, Myrinet, Optical fiber, or Wireless LAN network. In addition, the network topology of network 1000 may also vary. Communication between the front end node 200 and the compute nodes 300a to 300d may be accomplished using an application programming interface (API), such as the Message Passing Interface (MPI), that allows computers to communicate with one another.

FIG. 2 is a detailed block diagram depicting the internal architecture of an example front end node.

As shown in FIG. 2, front end node 200 includes central processing unit (CPU) 202 which interfaces with computer bus 201. Also interfacing with computer bus 201 are fixed disk 208 (e.g., a hard disk or other nonvolatile storage medium), network interface 203, keyboard interface 204, mouse interface 205, random access memory (RAM) 206 for use as a main run-time transient memory, read only memory (ROM) 207, and work distribution module 214. Front end node 200 uses network interface 203 to interface to network 1000.

RAM 206 interfaces with computer bus 201 so as to provide information stored in RAM 206 to CPU 202 during execution of the instructions in software programs, such as an operating system, application programs, the work distribution module, and device drivers. More specifically, CPU 202 first loads computer-executable process steps from fixed disk 208, or another storage device, into a region of RAM 206. CPU 202 can then execute the stored process steps from RAM 206 in order to execute the loaded computer-executable process steps. Data, or other information, is stored in RAM 206 and the data is accessed by CPU 202 during the execution of the computer-executable software programs, to the extent that such software programs have a need to access and/or modify the data.

As also shown in FIG. 2, fixed disk 208 is an example of a computer-readable storage medium that stores computer-executable process steps for operating system 209, and application programs 210, such as moving image manipulation programs. Fixed disk 208 also stores computer-executable process steps for device drivers for software interface to devices, such as input device drivers 211, output device drivers 212, and other device drivers 213. Image files 220 and other files 221 are stored for distribution to compute nodes for processing.

Fixed disk 208 also includes a work queue 222 that receives and stores work to be processed. The work stored in work queue 222 is distributed and processed in the order it is received.

Work distribution module 214 comprises computer-executable process steps stored on a computer-readable storage medium for execution by a processor, such as CPU 202, to extract tasks from work that is received by front end node 200, and to distribute those tasks to compute nodes for processing. Work distribution module 214 generally comprises an extraction module 215, a distribution module 216, a receiving module 217, a determination module 218, and a job load adjuster module 219. These modules are described in greater detail below in connection with FIG. 4.

The computer-executable process steps for work distribution module 214 may be configured as part of operating system 209, as part of an output device driver, such as a work distribution driver, or as a stand-alone application program. Work distribution module 214 may also be configured as a plug-in or dynamic link library (DLL) to the operating system, a device driver, or an application program.

FIG. 3 is a detailed block diagram depicting the internal architecture of an example compute node.

As shown in FIG. 3, a compute node includes central processing unit (CPU) 302 which interfaces with computer bus 301. Also interfacing with computer bus 301 are fixed disk 308 (e.g., a hard disk or other nonvolatile storage medium), network interface 303, keyboard interface 304, mouse interface 305, random access memory (RAM) 306 for use as a main run-time transient memory, read only memory (ROM) 307, and a work processing module 314. Network interface 303 is provided to interface to network 1000.

RAM 306 interfaces with computer bus 301 so as to provide information stored in RAM 306 to CPU 302 during execution of the instructions in software programs, such as an operating system, application programs, a work processing module, and device drivers. More specifically, CPU 302 first loads computer-executable process steps from fixed disk 308, or another storage device, into a region of RAM 306. CPU 302 can then execute the stored process steps from RAM 306 in order to execute the loaded computer-executable process steps. Data, or other information, is stored in RAM 306 and the data is accessed by CPU 302 during the execution of the computer-executable software programs, to the extent that such software programs have a need to access and/or modify the data.

As also shown in FIG. 3, fixed disk 308 is an example of a computer-readable storage medium that stores computer-executable process steps for operating system 309, and application programs 310, such as moving image manipulation programs. Fixed disk 308 also stores computer-executable process steps for device drivers for software interface to devices, such as input device drivers 311, output device drivers 312, and other device drivers 313.

Work processing module 314 comprises computer-executable process steps stored on a computer-readable storage medium for execution by a processor, such as CPU 302, to receive and execute tasks. Work processing module 314 generally comprises a receiving module 315, an execution module 316, and an output module 317. These modules are described in greater detail below in connection with FIG. 4.

The computer-executable process steps for work processing module 314 may be configured as part of operating system 309, as part of an output device driver, such as a work processing driver, or as a stand-alone application program. Work processing module 314 may also be configured as a plug-in or dynamic link library (DLL) to the operating system, a device driver, or an application program.

FIG. 4 illustrates example architecture for a work distribution module in a front end node and example architecture for a work processing module in a compute node, and interaction therebetween.

In particular, FIG. 4 illustrates one example of the work distribution module 214 in which the sub-modules of work distribution module 214 are stored in fixed disk 208. The example work distribution module 214 is constructed to distribute work from the front end node 200 to the compute nodes 300a to 300d over a network 1000 for processing. As shown in FIG. 4, the work distribution module 214 comprises an extraction module 215, a distribution module 216, a receiving module 217, a determination module 218, and a job load metric adjuster module 219. Each of these modules are computer-executable software code or process steps executable by a processor, such as CPU 202, and are stored on a computer-readable storage medium, such as fixed disk 208 or RAM 206. More or fewer modules may be used than shown in FIG. 4, and other architectures are possible.

As shown in FIG. 4, the work distribution module 214 includes an extraction module 215 constructed to extract tasks from the work received by front node 200, where the number of tasks corresponds to the number of nodes, and where the tasks are sized based on a job load metric of each of the nodes. The extraction module 215 is further constructed to extract tasks from the work received by front node 200, where the tasks are sized based on an adjusted job load metric of each of the compute nodes.

Work distribution module 214 also includes a distribution module 216 constructed to distribute the tasks to at least some of the nodes for processing.

In one example embodiment, the work available for distribution is dynamic in nature such that the amount of work available for distribution is received in work queue 222 on an ongoing basis, and where the distribution module 216 is further constructed to distribute the work available for distribution in the order received. For example, as shown in FIG. 1, the front end node 200 receives a moving image 100 via the Internet (or intranet) on an ongoing basis, and stores the received frames 101 to 105 in work queue 222. The front end node 200 distributes the frames 101 to 105 from the work queue 222, for processing by the compute nodes 300a to 300d, in the order received.

In another example embodiment, the distribution module 216 of front end node 200 is further constructed to transmit a qualification task for execution by each node, where the qualification task measures capabilities of core elements in a node, where an influence factor is assigned to each core element, and where the influence factor is used to adjust the weight of the core element on a capability measurement of the node. Qualification tasks are described in more detail below, in connection with FIG. 5.

The work distribution module 214 includes a receiving module 217 constructed to receive results outputted by the compute nodes upon execution of their assigned tasks.

Work distribution module 214 further includes a determination module 218 constructed to determine a time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task, and to determine whether such time difference exceeds a predefined threshold.

In addition, work distribution module 214 includes a job load metric adjuster module 219 constructed to adjust the job load metric of at least some of the nodes in response to a determination that the time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task exceeds a predefined threshold.

In one example embodiment, the job load metric adjuster module 219 is further constructed to calculate an initial job load metric for each compute node using the results from a qualification task.

In another example embodiment, the job load metric adjuster module 219 is further constructed to leave the job load metric unadjusted in response to a determination that the time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task does not exceed a predefined threshold.

FIG. 4 also illustrates an example work processing module 314 in which sub-modules of the work processing module 314 are stored in fixed disk 308. Specifically, FIG. 4 illustrates one example of the work processing module 314 constructed to receive tasks from front end node 200 over network 1000, and to execute the received tasks. As shown in FIG. 4, the work processing module 314 comprises a receiving module 315, an execution module 316, and an output module 317. Each of these modules are computer-executable software code or process steps executable by a processor, such as CPU 302, and are stored on a computer-readable storage medium, such as fixed disk 308 or RAM 306. More or fewer modules may be used than shown in FIG. 4, and other architectures are possible.

As shown in FIG. 4, the work processing module 314 includes a receiving module 315 constructed to receive tasks from front end node 200 for processing.

The work processing module 314 also includes an execution module 316 constructed to execute tasks received from the front end node 200.

The work processing module 314 also includes an output module 317 constructed to output the results of the executed tasks to the front end node 200.

FIG. 5 is a flowchart illustrating an example process for explaining task allocation.

Briefly, in FIG. 5, work is distributed amongst a plurality of nodes in a distributed computing environment. A first plurality of tasks is extracted from the work, where the number of tasks in the first plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the first plurality of tasks are sized based on a job load metric of each of the nodes. The first plurality of tasks is distributed to at least some of the nodes for processing. A determination is made whether a time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task exceeds a predefined threshold. The job load metric of at least some of the nodes is adjusted in response to a determination that the time difference exceeds the predefined threshold. A second plurality of tasks is extracted from the work, where the number of tasks in the second plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the second plurality of tasks are sized based on the adjusted job load metric of each of the nodes. The second plurality of tasks is distributed to at least some of the nodes for processing.

In more detail, the process in the front end node 200 starts with step 501, where the front end node 200 receives work to be processed by the compute nodes. As shown in FIG. 1, the front end node 200 receives work via a network, such as the Internet (or intranet). The front end node 200 may receive work from a single source or from multiple sources, and each source may transmit different types of work to be processed. For example, one source may transmit multiple time-sequential 4K-sized frames of a video stream for post-processing, another source may transmit a stream of images from security cameras for face detection purposes, and third source may transmit a high-definition video stream for special effects processing. Each stream of work is collected in a work queue 222, and is distributed by the front end node 200 to the compute nodes 300a to 300d on a first-in, first-out (FIFO) basis.

In step 502, the distribution module 216 of front end node 200 transmits a qualification task for execution by each compute node 300a to 300d, where the qualification task measures capabilities of core elements in a compute node. In this example, six different capabilities are measured: the type of network interconnect (I.sub.E), the nature and power of the CPU (C.sub.E), the nature and power of the GPU (G.sub.E), the capabilities of storage devices (S.sub.E), the capabilities of the system memory (M.sub.E), and system bus performance (B.sub.E). More precisely, the type of network interconnect I.sub.E may distinguish between Ethernet, Infiniband, Myrinet, Optical fiber, or copper wire. The nature and power of the CPU C.sub.E may be determined using the clock frequency, cache size, and processing cores. Similarly, the nature and power of the GPU G.sub.E may be determined using the clock frequency, device memory size, and processing cores. The capabilities of the storage devices S.sub.E may be established based on storage device capacity, buffer size, and data throughput. System memory M.sub.E capabilities may be based on the clock frequency and storage capacity of the system memory. Finally, the performance of the system bus B.sub.E may be based on the maximum throughput of the bus.

The present disclosure contemplates modifying the qualification task to measure capabilities of other core elements in addition to the core elements described above, thereby allowing measurement of core elements that are relevant to a particular type of work.

In one example embodiment, the distribution module 216 of front end node 200 tailors the influence factors in correspondence to the nature of the work. For example, a compute environment requiring a large number of CPU-intensive transactions would typically place most of the demand on the CPU and the system memory, followed by network bandwidth and bus speed, with the storage device and the GPU having the least amount of influence. As discussed in more detail below in connection with step 505, in this example, C.sub.E and M.sub.E would be assigned the highest influence factors, with I.sub.E and B.sub.E having smaller influence factors, and with S.sub.E and G.sub.E having the lowest influence factors. According to the embodiment, the value of each influence factor is set manually. However, the present disclosure also contemplates determining and setting each influence factor automatically using heuristic methods, such as machine learning and data mining.

In step 503, the receiving module 217 of front end node 200 receives the results from the executed qualification task, from each compute node 300a to 300d. The qualification task results are stored for use in calculating a job load metric for the compute nodes, as discussed more fully in step 505. The following table illustrates example core element capability measurements that were generated upon execution of a qualification task by compute nodes 300a to 300d:

TABLE-US-00001 TABLE 1 Example Core Element Measurement Node I.sub.E C.sub.E G.sub.E S.sub.E M.sub.E B.sub.E Node 1 100 Mbps 3 GHz/512K/1 64/128M/1 GHz 0.2 Gb/8M/200 GB 800 MHz/1 GB 0.5 Gbps Node 2 1 Gbps 1.2 GHz/1M/4 32/128M/1 GHz 0.15 Gb/16M/500 GB 1000 MHz/4 GB 2 Gbps Node 3 1 Gbps 1.1 GHz/2M/2 256/1G/1.1 GHz 0.2 Gb/32M/750G 1200 MHz/8 GB 1 Gbps Node 4 20 Gbps 2 GHz/2M/6 312/1.9G/1 GHz 0.3 Gb/64M/400 GB 1600 MHz/16 GB 1 Gbps

In step 504, the job load metric adjuster module 219 of front end node 200 ranks the core elements in proportion to one another, as shown in Table 2.

The following table illustrates an example proportionate ranking of core elements C.sub.E, M.sub.E, I.sub.E, and B.sub.E using the core element capability measurements determined in steps 502 and 503, as shown in Table 1.

TABLE-US-00002 TABLE 2 Example Proportionate Rankings for Core Elements Node C.sub.E M.sub.E I.sub.E B.sub.E Node 1 0.10 0.05 0.05 0.16 Node 2 0.30 0.20 0.20 0.40 Node 3 0.20 0.30 0.20 0.22 Node 4 0.40 0.45 0.55 0.22

In step 505, the job load adjuster module 219 of front end node 200 computes a capability metric (CM) for each core element using the proportionate ranking data for the core elements shown in Table 2. The capability metric (CM) is determined as follows: CM(X.sub.E)=X.sub.E.times.I.sub.F(X.sub.E) Equation 1

where CM is capability metric, X.sub.E is the core element proportionate ranking, and I.sub.F is the influence factor assigned to the core element.

Using Equation 1, the front end node 200 determines a capability metric for core elements C.sub.E, M.sub.E, I.sub.E, and B.sub.E using the core element data from Table 2 and an influence factor (I.sub.F). For simplicity purposes, the present example omits core element data for S.sub.E and G.sub.E from Table 2 in order to deemphasize their role in the allocation of tasks in a CPU-intensive environment, where the capabilities of S.sub.E and G.sub.E would typically be less significant.

In one example embodiment, an influence factor is assigned to each core element, where the influence factor is used to adjust the weight of the core element on a capability measurement of the compute node.

The influence factor is tailored to allow the front end node 200 to assign appropriate weights to the core elements, thereby allowing the front end node 200 to manipulate the amount of influence a particular core element has on the capability metric and, ultimately, the job load metric, in correspondence to the nature of the work. For example, a compute environment requiring a large number of CPU-intensive transactions would typically place most of the demand on the CPU and the system memory. Additional factors, such as network bandwidth and bus speed, would also play a role in such a scenario, since the ability to transfer data quickly between processors and compute nodes is needed. In such an environment, the highest influence factor would be assigned to the CPU (C.sub.E) and system memory (M.sub.E) core elements, followed by the network interconnects (I.sub.E) core element and bus speed (B.sub.E). The following table shows example capability metric values for core elements C.sub.E, M.sub.E, I.sub.E, and B.sub.E, that were calculated using Equation 1 and the core element data from Table 2, where core elements C.sub.E and M.sub.E were assigned an influence factor of 0.3, core elements I.sub.E and B.sub.E were assigned influence factors 0.2 and 0.15, respectively.

TABLE-US-00003 TABLE 3 Example Capability Metric for Core Elements Node CM (C.sub.E) CM (M.sub.E) CM (I.sub.E) CM (B.sub.E) Node 1 0.03 0.015 0.01 0.024 Node 2 0.09 0.06 0.04 0.06 Node 3 0.06 0.09 0.04 0.033 Node 4 0.12 0.135 0.11 0.033

Once the capability metrics have been determined, in step 506, the front end node 200 determines the capacity utilization for core elements C.sub.E, M.sub.E, I.sub.E, and B.sub.E using the following equation:

.function..function..times..function..times..times..times. ##EQU00001##

where CU is the capacity utilization, CM is the capability metric, and X.sub.E is the core element proportionate ranking.

The front end node 200 uses the capacity utilization metrics to determine a job load metric for each compute node. The job load metric of each compute node is used to allocate the tasks to the compute nodes, where the allocation is performed so as to achieve near full utilization of the core elements. As discussed above, an environment having a large number of CPU-intensive transactions would typically be heavily dependent on the CPU (C.sub.E), system memory (M.sub.E), network interconnects (I.sub.E), and bus speed (B.sub.E). Thus, the capacity utilization metric would aid in allocating the tasks so that the C.sub.E, M.sub.E, I.sub.E, and B.sub.E core elements are utilized at or near 100 percent. The following table shows example capacity utilization metrics for core elements C.sub.E, M.sub.E, I.sub.E, and B.sub.E that were calculated using Equation 2 in combination with capability metric data from Table 3.

TABLE-US-00004 TABLE 4 Example Capacity Utilization of Core Elements Node CU (C.sub.E) % CU (M.sub.E) % CU (I.sub.E) % CU (B.sub.E) % Node 1 10 5 5 16 Node 2 30 20 20 40 Node 3 20 30 20 22 Node 4 40 45 55 22

In step 507, the front end node 200 calculates a job load metric for each compute node 300a to 300d by using the results from the executed qualification task. Specifically, a job load metric is approximated using the following equation:

.apprxeq..times..function..times..function..times..times. ##EQU00002##

where JL is the job load metric, CU is the capacity utilization, X.sub.E is the core element proportionate ranking, I.sub.F is the influence factor assigned to the core element, and where the summation is taken over "6" since, in this example, there are six core elements.

A job load metric is calculated for each compute node 300a to 300d using Equation 3 and the capacity utilization values that were determined in step 506. In particular, the job load metrics for the compute nodes 300a to 300d are used by the front end node 200 to size the tasks extracted from the work.

In step 508, the extraction module 215 of front end node 200 extracts a first plurality of tasks, where the number of tasks in the first plurality of tasks is selected in correspondence to the number of nodes, and where sizes of the tasks in the first plurality of tasks are sized based on a job load metric of each of the compute nodes. In one example, the extraction module 215 of front end node 200 divides the work into four tasks, one for each compute node 300a to 300d. The sizes of the divided tasks correspond to the job load metric of the compute node assigned to execute that task. In other words, the size of a task assigned to compute node 300a would be based on a job load metric that was determined for compute node 300a. Similarly, the sizes of tasks assigned to compute nodes 300b, 300c and 300d, is based on a job load metric that was determined for compute nodes 300b, 300c and 300d, respectively.

In step 509, the distribution module 216 of front end node 200 distributes the first plurality of tasks to at least some of the nodes for processing.

In step 510, the receiving module 217 of front end node 200 receives the executed task results that are outputted by output module 317 of compute nodes 300a to 300d. The front end node 200 may then store the task results in a database or may send the results to another server 106 for further processing and display, as depicted in FIG. 1.

In step 511, the determination module 218 of front end node 200 determines whether a time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task exceeds a predefined threshold. If the time difference exceeds the predefined threshold, the front end node 200 proceeds to step 512. Otherwise, if the time difference does not exceed the predefined threshold, the front end node proceeds to step 508, where the extraction module 215 of front end computer 200 continues processing work from work queue 222.

As suggested above, the time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task is a measurement that is used to gauge how far apart in time the responses are for the best performing compute node and for the worst performing compute node. Ideally, the time difference between responses from the best performing compute node and the worst performing compute node should be minimal, which would indicate that all of the compute nodes are being utilized at near full potential. However, a significant time difference between responses from the best performing node and the worst performing node would suggest that the best performing compute node may be under-worked, as indicated by its fast response, or that the worst performing compute node may be over-worked, as indicated by its slow response.

The predefined threshold is used to determine whether the time difference between responses from the best performing compute node and the worst performing compute node falls within an acceptable duration. A time difference that exceeds the predefined threshold will trigger an adjustment of the job load metric for the compute nodes. Therefore, setting the predefined threshold too high may result in the job load metric not being adjusted, and may therefore result in a sub-optimal utilization of the compute nodes. For example, if the predefined threshold is defined as 2.0 seconds, and the time difference between the first node to respond and the last node to respond never exceeds 2.0 seconds, then adjustment of the job load metric will not be triggered. In this example, the front end node 200 will size the tasks using the unadjusted job load metric each time tasks are extracted from the work, which may result in a sub-optimal sizing of the tasks. Alternatively, setting the predefined threshold too low may result in the job load metric being adjusted too often. Since adjusting the job load metric requires the front end node 200 to perform additional operations, frequent adjustment of the job load metric may disrupt efficient operation of the computing cluster. For example, if the predefined threshold is defined as 0.1 seconds, and the time difference between the first node to respond and the last node to respond frequently exceeds 0.1 seconds, then adjustment of the job load metric would be triggered each time tasks are extracted from the work. Such a situation is ordinarily not desirable because it affects the efficient operation of the computing cluster. Thus, setting the predefined threshold value to an acceptable value is helpful in maintaining efficiency. According to the embodiment, the threshold is set manually. However, the present disclosure also contemplates determining and setting the predefined threshold automatically using heuristic methods, such as machine learning and data mining.

In steps 512 and 513, the job load adjuster module 219 of front end node 200 adjusts the job load metric of at least some of the nodes in response to a determination that the time difference between a response from a node that is first to complete its task and a response from a node that is last to complete its task exceeds the predefined threshold.

The description continues in the full USPTO document.