Source: https://patents.google.com/patent/US10310902
Timestamp: 2019-06-19 12:20:57
Document Index: 647486725

Matched Legal Cases: ['Application No. 61', 'Application No. 61', 'Application No. 61', 'Application No. 61', 'Application No. 61', 'Application No. 401']

US10310902B2 - System and method for input data load adaptive parallel processing - Google Patents
US10310902B2
US10310902B2 US16/226,502 US201816226502A US10310902B2 US 10310902 B2 US10310902 B2 US 10310902B2 US 201816226502 A US201816226502 A US 201816226502A US 10310902 B2 US10310902 B2 US 10310902B2
US16/226,502
US20190114209A1 (en
2012-06-08 Priority to US201261657708P priority
2012-07-19 Priority to US201261673725P priority
2012-11-02 Priority to US201261721686P priority
2012-11-16 Priority to US201261727372P priority
2014-04-24 Priority to US14/261,384 priority patent/US9262204B2/en
2014-10-23 Priority to US14/521,490 priority patent/US20180349969A9/en
2016-02-12 Priority to US15/042,159 priority patent/US9400694B2/en
2016-06-16 Priority to US15/183,860 priority patent/US9465667B1/en
2016-09-23 Priority to US15/273,731 priority patent/US20170109208A1/en
2018-03-23 Priority to US15/933,724 priority patent/US10061615B2/en
2018-06-21 Priority to US16/014,674 priority patent/US10133600B2/en
2018-09-28 Priority to US16/145,632 priority patent/US10310901B2/en
2018-12-19 Priority to US16/226,502 priority patent/US10310902B2/en
2018-12-19 Application filed by Mark Henrik Sandstrom filed Critical Mark Henrik Sandstrom
2019-04-18 Publication of US20190114209A1 publication Critical patent/US20190114209A1/en
2019-06-04 Publication of US10310902B2 publication Critical patent/US10310902B2/en
This application is a continuation of U.S. application Ser. No. 16/145,632 filed Sep. 28, 2018, which is a continuation of U.S. application Ser. No. 16/014,674 filed Jun. 21, 2018 (now U.S. Pat. No. 10,133,600), which is a continuation of U.S. application Ser. No. 15/933,724 filed Mar. 23, 2018 (now U.S. Pat. No. 10,061,615), which is a continuation of U.S. application Ser. No. 15/273,731 filed Sep. 23, 2016, which is a continuation of U.S. application Ser. No. 15/183,860 filed Jun. 16, 2016 (now U.S. Pat. No. 9,465,667), which is a divisional of U.S. application Ser. No. 15/042,159 filed Feb. 12, 2016 (now U.S. Pat. No. 9,400,694), which is a continuation of U.S. application Ser. No. 14/261,384 filed Apr. 24, 2014 (now U.S. Pat. No. 9,262,204), which is a continuation of U.S. application Ser. No. 13/684,473 filed Nov. 23, 2012 (now U.S. Pat. No. 8,789,065), which claims the benefit of the following provisional applications:
U.S. Provisional Application No. 61/657,708, filed Jun. 8, 2012;
U.S. Provisional Application No. 61/673,725, filed Jul. 19, 2012;
U.S. Provisional Application No. 61/721,686, filed Nov. 2, 2012; and
U.S. Provisional Application No. 61/727,372, filed Nov. 16, 2012.
U.S. application Ser. No. 16/014,674 is also a continuation of U.S. Utility application Ser. No. 14/521,490, filed Oct. 23, 2014, which is a continuation of U.S. Utility application Ser. No. 13/297,455, filed Nov. 16, 2011, which claims the benefit of U.S. Provisional Application No. 61/556,065, filed Nov. 4, 2011.
U.S. Utility application Ser. No. 13/184,028, filed Jul. 15, 2011;
U.S. Utility application Ser. No. 13/270,194, filed Oct. 10, 2011 (now U.S. Pat. No. 8,490,111); and
U.S. Utility application Ser. No. 13/277,739, filed Nov. 21, 2011 (now U.S. Pat. No. 8,561,076)
All above identified applications are hereby incorporated by reference in their entireties for all purposes.
For brevity: ‘application (program)’ is occasionally written in as ‘app’, ‘instance’ as ‘inst’ and ‘application-task/instance’ as ‘app-task/ins’.
Identifiers such as ‘master’ and ‘worker’ tasks or processing stages are not used here in a sense to restrict the nature of such tasks or processing; these identifiers are here used primarily to illustrate a possible, basic type of distribution of workloads among different actors. For instance, the entry stage processing system may host, for a given application, simply tasks that pre-process (e.g. qualify, filter, classify, format, etc.) the RX data units and pass them to the worker stage processing systems as tagged with the pre-processing notations, while the worker stage processor systems may host the actual master (as well as worker) actors conducting the main data processing called for by such received data units. Generally, a key idea of the presented processing system and IO architecture is that the worker stages of processing—where bulk of the intra-application parallel and/or pipelined processing typically is to occur, providing the performance gain of using parallel task instances and/or pipelined tasks to lower the processing latency and improve the on-time IO throughput— receive their input data units as directed to specific destination app-task instances, while the external parties are allowed to communicate with a given application program hosted on a system 1 through a single, constant contact point (the ‘master’ task hosted on the entry stage processor, possibly with its specified instance).
While the processing of any given application (server program) at a system 1 is normally parallelized and/or pipelined, and involves multiple tasks (many of which tasks and instances thereof can execute simultaneously on the manycore arrays of the processors 300), the system enables external parties to communicate with any such application hosted on the system 1 without having to know about any specifics (incl. existence, status, location) of their internal tasks or parallel instances thereof. As such, the incoming data units to the system 1 are expected to identify just their destination application (and where it matters, the application instance number), rather than any particular task within it. Moreover, the system enables external parties to communicate with any given application hosted on a system 1 through any of the network ports 10, 50 without knowing whether or at which cores any instance of the given application task (app-task) may be executing at any time. Furthermore, the architecture enables the aforesaid flexibility and efficiency through its hardware logic functionality, so that no system or application software running on the system 1 needs to either be aware of whether or where any of the instances of any of the app-tasks may be executing at any given time, or through which port any given inter-task or external communication may have occurred or be occurring. Thus the system 1, while providing a highly dynamic, application workload adaptive usage of the system processing and communications resources, allows the software running on and/or remotely using the system to be designed with a straightforward, abstracted view of the system: the software (both the server programs hosted on a system 1 as well as clients etc. remote agents interacting with such programs hosted on the system) can assume that all applications (as well all their tasks and instances thereof) hosted on by the given system 1 are always executing on their virtual dedicated processor cores within the system. Also, where useful, said virtual dedicated processors can also be considered by software to be time-share slices on a single (very high speed) processor. The architecture thereby enables achieving, at the same time, both the vital application software development productivity (simple, virtual static view of the actually highly dynamic processing hardware) together with high program runtime performance (scalable parallel program execution with minimized overhead) and resource efficiency (adaptively optimized resource allocation) benefits. Techniques enabling such benefits of the architecture are described in the following through more detailed technical study of the system 1 and its sub systems.
Stage# t
(below) 0 1 2 3
index # --
internal (lower Target core type
index value App-inst ID # (identifies the (e.g., 0 denotes CPU,
signifies more app-inst-specific memory 950 1 denotes DSP, and
urgent app-inst) within the memory array 850) 2 denotes GPU, etc.)
1. A system, executing on at least one of hardware logic and software logic executing on a plurality of processors, for hosting a plurality of application programs, the system comprising:
a plurality of processing data input buffers, each input buffer of the plurality of processing data input buffers queuing data for a corresponding instance of one program of the plurality of application programs, wherein each program of the plurality of programs comprises a plurality of instances;
an array of cores of computing capacity;
a first subsystem configured to allocate the array of cores of computing capacity among the plurality of application programs, wherein
allocating comprises allocating the array of cores of computing capacity based at least in part on
a respective volume of processing data at each buffer of the plurality of processing data input buffers, and
a respective processing quota of each application program of at least a portion of the plurality of application programs,
allocating comprises allocating more than one core of the array of cores of computing capacity to at least one of the plurality of application programs;
a second subsystem configured to
assign, for each program of the plurality of application programs, each core allocated to the respective program to a different instance of the plurality of instances of the respective program, wherein
the assigning results in assignment of a plurality of selected instances of the plurality of instances of the plurality of application programs, the plurality of instances comprising one or more executable instances of each program of the plurality of application programs, wherein
a number of the plurality of selected instances is fewer than a maximum number of the plurality of instances, and
the plurality of selected instances is selected based at least in part on respective volumes of processing data available for each instance of the plurality of instances of the respective program at the portion of the plurality of data input buffers queuing data for the respective program, and
according to the assigning, control connectivity between the plurality of processing data input buffers and the array of cores; and
a third subsystem configured, according to the controlling, to establish direct data access from each input buffer of at least a subset of the plurality of processing data input buffers to the respective core of the array of cores that is assigned to a given corresponding instance of the program for which the respective input buffer is queuing data;
wherein the array of cores is periodically allocated by the first subsystem and assigned by the second subsystem based at least in part on changes in respective volumes of processing data associated with each program of the plurality of application programs.
2. The system of claim 1, wherein controlling connectivity comprises providing, for each core of the array of cores of computing capacity, an identification of an instance of a program of the plurality of application programs assigned for execution on the respective core.
controlling connectivity comprises providing an input buffer selection configuration for at least one multiplexer of a plurality of core-specific multiplexers for causing connection of input data from a particular input buffer of the plurality of processing data input buffers to a particular core of the array of cores of computing capacity, wherein
the at least one multiplexer is specific to the particular core.
4. The system of claim 1, wherein allocating comprises, for at least one program of the plurality of application programs:
quantifying the respective volume of processing data queued for the respective program in one or more input buffers corresponding to the respective program as a count of processing data units; and
allocating one or more cores to the respective program based at least in part on the count of processing data units.
5. The system of claim 1, wherein the respective processing quota comprises an entitlement held by each program of the portion of the plurality of application programs.
6. The system of claim 5, wherein a first plurality of cores of the array of cores of computing capacity is allocated so that an expression of demand for the array of cores of computing capacity by each of the portion of the plurality of application programs is assured to be met at least up to the entitlement of the respective program.
7. The system of claim 6, wherein, for each program of the plurality of application programs, the respective expression of demand by the respective program is based at least in part on a volume of processing data at the portion of the plurality of processing data input buffers queuing data for the respective program.
8. The system of claim 1, wherein, for at least one program of the plurality of application programs, a maximum number of the one or more executable instances of the respective program includes an upper limit of a number of concurrently executable instances of the at least one program.
9. A method, implemented on hardware logic and/or on software logic executing on processors, for executing a plurality of application programs, the method comprising:
on each of a plurality of data input buffers, queuing respective data for a corresponding program of the plurality of application programs;
periodically allocating an array of units of computing capacity among the plurality of application programs, wherein allocating comprises, for each period of a plurality of allocation periods,
identifying expressions of demand for the array of units of computing capacity by each program of the plurality of application programs, wherein
the respective expression of demand for a given program of the plurality of application programs is based at least in part on quantifying the respective data queued for the given program in one or more input buffers of the plurality of data input buffers corresponding to the given program as a count of processing data units, and
allocating the array of units of computing capacity based at least in part on the expressions of demand of the plurality of application programs; and
for each period of a plurality of allocation periods, responsive to the allocating,
assigning, for each program of the plurality of application programs, one or more instances of the respective program for execution on a respective one or more units of the array of units of computing capacity, wherein
the one or more instances are assigned according to i) a respective number of the units of the array of units of computing capacity allocated to the respective program, and ii) execution priorities among instances of the respective program, and
according to the assigning, establishing direct data access from each buffer of at least a subset of the plurality of data input buffers to the cores respective unit of the array of units of computing capacity that is assigned a given corresponding instance of the respective program for which the respective data buffer is queuing input data, wherein
establishing direct data access comprises enabling the given corresponding instance to access and process the respective input data at the respective data buffer.
10. The method of claim 9, wherein, for at least one program of the plurality of application programs:
queuing the respective data comprises queuing respective amounts of data for individual instances of the respective program; and
assigning comprises determining the execution priorities based at least in part on the respective amounts of data queued to the individual instances of the respective program.
11. The method of claim 9, wherein, for each period of the plurality of allocation periods and for each program of at least a portion of the plurality of application programs, the array of units of computing capacity is allocated based further in part on a respective entitlement to the array of units of computing capacity held by the respective program, wherein
periodically allocating comprises ensuring the respective expressions of demand for the units of computing capacity at least up to the entitlement are met for each program of the portion of the plurality of application programs.
12. The method of claim 11, wherein, for each period of at least a portion of the plurality of allocation periods, after ensuring that the respective expressions of demand for the units of computing capacity at least up to entitlements are met for each of the portion of the plurality of application programs, remaining units of the array of units of computing capacity are allocated based further in part on remaining unmet demand for the remaining units by one or more programs of the plurality of application programs until i) the demand by each of the one or more programs is met, or ii) all units of computing capacity are allocated.
13. The system of claim 1, further comprising a fourth subsystem configured to, prior to assigning the plurality of selected instances,
for each program of the plurality of application programs, selecting at least one instance, wherein
selecting is based at least in part on relative readiness for execution among the plurality of instances of the respective program, and
a different instance of the respective program is selected for each core allocated to the respective program.
14. A system, executing on at least one of hardware logic and software logic executing on a plurality of processors, for hosting a plurality of application programs, the system comprising:
a plurality of data buffers, each buffer of the plurality of data buffers queuing input data for a respective program of the plurality of application programs, wherein
the respective program is configured to access and process the input data during execution of the respective program;
a first subsystem configured to allocate the plurality of processing units among the plurality of application programs, wherein allocating comprises
allocating the plurality of processing units based at least in part on a respective volume of input data at each buffer of the plurality of data buffers,
allocating, for at least some time periods, more than one processing unit of the plurality of processing units to at least one of the plurality of application programs;
a second subsystem configured to, for each program of the plurality of application programs, select one or more instances of a plurality of instances of the respective program, wherein
selecting is based at least in part on respective volumes of input data available for each instance of the plurality of instances of the respective program, and
a different instance of the respective program is selected for each processing unit allocated to the respective program;
a third subsystem configured to assign, for each program of the plurality of application programs, each processing unit allocated to the respective program to a different instance of the one or more instances of the respective program in accordance with the selecting; and
a fourth subsystem configured, according to the assigning, to establish direct data access from each buffer of at least a subset of the plurality of data buffers to the respective processing unit of the plurality of processing units that is assigned a given corresponding instance of the respective program for which the respective data buffer is queuing input data, wherein
establishing direct data access comprises enabling the given corresponding instance to access and process the respective input data at the respective data buffer;
wherein the plurality of processing units is periodically allocated by the first subsystem, selected by the second subsystem, and assigned by the third subsystem based at least in part on changes in respective volumes of input data associated with each program of the plurality of application programs.
15. The system of claim 14, wherein allocating comprises allocating the plurality of processing units based further in part on a respective entitlement of each program of at least a portion of the plurality of application programs to processing resources of the plurality of processing units.
16. The system of claim 14, wherein selecting based at least in part on respective volumes of input data comprises selecting based at least in part on quantifying a number of queued items of the respective data queued for the respective program in one or more buffers of the plurality of data buffers corresponding to the respective program.
a plurality of program instance specific multiplexers, each program instance specific multiplexer being specific to one instance of one of the plurality of application programs;
wherein establishing direct data access comprises, for each processing unit of at least a portion of the plurality of processing units, controlling, using the program instance specific multiplexer of the instance assigned to the respective processing unit, connectivity between the respective processing unit and the program specific data buffer of a program of the plurality of application programs corresponding to the instance assigned to the respective processing unit.
US16/226,502 2010-12-30 2018-12-19 System and method for input data load adaptive parallel processing Active US10310902B2 (en)
US201261657708P true 2012-06-08 2012-06-08
US201261673725P true 2012-07-19 2012-07-19
US201261721686P true 2012-11-02 2012-11-02
US201261727372P true 2012-11-16 2012-11-16
US15/933,724 US10061615B2 (en) 2012-06-08 2018-03-23 Application load adaptive multi-stage parallel data processing architecture
US16/145,632 US10310901B2 (en) 2018-09-28 System and method for input data load adaptive parallel processing
US16/226,502 US10310902B2 (en) 2018-12-19 System and method for input data load adaptive parallel processing
US16/145,632 Continuation US10310901B2 (en) 2010-12-30 2018-09-28 System and method for input data load adaptive parallel processing
US20190114209A1 US20190114209A1 (en) 2019-04-18
US10310902B2 true US10310902B2 (en) 2019-06-04
US20120266176A1 (en) 2011-04-18 2012-10-18 Microsoft Corporation Allocating Tasks to Machines in Computing Clusters
US20160378538A1 (en) 2010-07-01 2016-12-29 Neodana, Inc. Partitioning processes across clusters by process type to optimize use of cluster specific configurations
US20170097838A1 (en) 2013-03-15 2017-04-06 Applied Micro Circuits Corporation Virtual appliance on a chip
US20160080201A1 (en) 2012-10-28 2016-03-17 Citrix Systems, Inc. Network offering in cloud computing environment
[#HADOOP-3445] Implementing core scheduler functionality in Resource Manager (V1) for Hadoop, Accessed May 18, 2018, 12 pages, https://issues.apache.org/jira/si/jira.issueviews:issue-html/HADOOP-3445/HADOOP-3445.html.
Cooper, Brian F. et al., Building a Cloud for Yahoo!, 2009, 9 pages, IEEE Computer Society Technical Committee on Data Engineering, https://www.researchgate.net/profile/Rodrigo_Fonseca3/publication/220282767_Building_a_Cloud_for Yahoo/links/0912f5109da99ddf6a000000/Building-a-Cloud-for-Yahoo.pdf.
Final Rejection issued in related U.S. Appl. No. 14/521,490 dated Nov. 14, 2018, 21 pages.
First Examination Report issued in IN Application No. 401/MUM/2011 on Nov. 9, 2018.
Ghodsi, Ali, et al., Dominant Resource Fairness: Fair Allocation of Multiple Resource Types, Proceedings of NSDI '11: 8th USENIX Symposium on Networked Systems Design and Implementation, Mar. 30, 2011, pp. 323-336.
Hindman, Benjamin, et al., Mesos: A Platform for Fine-Grained Resource Sharing in the Data Center, Proceedings of NSDI '11: 8th USENIX Symposium on Networked Systems Design and Implementation, Mar. 30, 2011, pp. 295-308.
Notice of Allowance issued in related U.S. Appl. No. 16/014,658 dated Aug. 29, 2018, 22 pages.
Notice of Allowance issued in related U.S. Appl. No. 16/014,674 dated Aug. 27, 2018, 21 pages.
Notice of Allowance issued in related U.S. Appl. No. 16/145,632 dated Dec. 13, 2018, 10 pages.
Wen et al., "Minimizing Migration on Grid Environments: an Experience on Sun Grid Engine" Journal of Information Technology and Applications, vol. 1, No. 4, pp. 297-304 (2007).
US8209395B2 (en) 2012-06-26 Scheduling in a high-performance computing (HPC) system