METHOD AND DEVICE FOR TASK ALLOCATION IN PARALLEL STREAMS BASED ON DYNAMIC RESOURCE STATE

TECHNICAL FIELD

[0001] Various aspects of this disclosure relate to methods and devices for task allocation in parallel streams based on a dynamic resource state.

BACKGROUND

[0002] When a large-scale stream computation is performed, parallelization may improve execution efficiency of arithmetic performed at the same time to make full use of computational resources such as central processing units (CPU(s)), memory, etc. A process may be a separate unit for resource allocation and scheduling, and a thread may be a basic unit for a CPU schedule. The thread can be a subset of the process. One thread may be generally focused on processing some specific task and may not independently own system resources except for only some resources for running, such as a program counter.

[0003] In an example application, one task manager, as a process, may manage one and more tasks. Each task can be a thread and can occupy a slot. Resources of each slot may be an average subset of resources of the whole task manager. No competition for the resources can be between slots. A user may set a number of slots in the task manager so that the user may determine at what granularity the tasks are segregated from each other. The number of slots can be set to a number of CPU cores available under the task manager, so that on average, each slot includes an average of one CPU core. Further in this application, an execution graph may be physical graph of a result of translating a logical graph for execution in a distributed runtime. Nodes of the execution graph can be tasks, and edges can indicate input/output relationships or partitions of data streams or data sets. [0004] In an example, every task manager may have 3 slots. When an execution graph is inputted to the task manager, four task managers can be allocated depending on a degree of parallelism, which can be four in phase A of the overall graph. Each pipeline may be executed on a separate one of the task managers. Each execution vertex of the execution graph can correspond to one slot of the separate one of the task managers. If there are not enough slots, a slot pool may send a request to a slot manager, and if the slot manager determines that there are enough resources in a cluster to meet this demand, it may send an assign command to the task manager. Then, the task manager can provide slots to the slot pool.

[0005] The above example may achieve parallel execution of the execution graph, which may be a stream. However, some slots in the task manager can still be free and not assigned to any tasks. The resources in the cluster may not be fully utilized. In addition, data interaction between execution vertices can cause increased overhead for thread switches and increased pressure on inter-process communication.

SUMMARY

[0006] Various embodiments concern a method for task allocation in parallel streams based on a dynamic resource state, the method including, based on computational resources for executing a job among a plurality of jobs to executed for service applications matching computational resources available in two or more of a plurality of nodes for executing the plurality of jobs, determining a similarity value between the computational resources for executing the job and the computational resources available in each of the two or more of the plurality of nodes, selecting, from the two or more of the plurality of nodes, a single node having a lowest similarity value between the computational resources available in the single node and the computational resources for executing the job, and allocating the job to the selected single node so that the single node executes a task corresponding to the job.

[0007] The method may further include, based on the computational resources for executing the job matching computational resources available in one of the plurality of nodes, allocating the job to the one of the plurality of nodes so that the one of the plurality of nodes executes the task.

[0008] The job queue may hold the plurality of jobs on a first-in-first-out (FIFO) basis.

[0009] The method may further include, based on the computational resources for executing the job not matching computational resources available in each of the plurality of nodes, waiting for the dynamic resource state to be updated, the dynamic resource state including information of the computational resources available in each of the plurality of nodes.

[0010] The method may further include, based on the computational resources for executing the job matching the computational resources available in the two or more of the plurality of nodes, selecting, from the two or more of the plurality of nodes, at least one node having a least number of tasks executing in the at least one node.

[0011] The method may further include, based on the selected at least one node including one node, allocating the job to the one node so that the one node executes the task.

[0012] The selecting the single node includes, based on the selected at least one node including multiple nodes, selecting, from the multiple nodes, the single node having the lowest similarity value.

[0013] The method may further include updating the dynamic resource state by decreasing the computational resources for executing the allocated job from computational resources available in a respective one of the plurality of nodes.

[0014] The method may further include allocating, to the allocated job, computational resources available in a respective one of the plurality of nodes.

[0015] The method may further include generating the task for the allocated job.

[0016] The method may further include executing the generated task.

[0017] The method may further include determining whether the executed task is complete.

[0018] The method may further include, based on the task being determined to be complete, releasing the allocated computational resources. [0019] The method may further include updating the dynamic resource state by increasing the computational resources available in the respective one of the plurality of nodes with the released computational resources.

[0020] The method may further include updating the dynamic resource state to include information of the determined similarity value between the computational resources for executing the job and the computational resources available in each of the two or more of the plurality of nodes.

[0021] The similarity value may be a difference value between a number of logical central processing units (CPUs) for executing the job and a number of logical CPUs available in a respective node among the plurality of nodes, added by a difference value between a size of logical memory for executing the job and a number of logical memory available in the respective node.

[0022] The computational resources for executing the job may include a number of logical central processing units (CPUs) and a size of logical memory, and the computational resources available in each of the plurality of nodes may include a number of idle logical CPUs and a size of idle logical memory.

[0023] A server may be configured to perform the method.

[0024] A computer program element may include program instructions, which, when executed by one or more processors, cause the one or more processors to perform the method.

[0025] A computer-readable medium may include program instructions, which, when executed by one or more processors, cause the one or more processors to perform the method.

BRIEF DESCRIPTION OF THE DRAWINGS

[0026] The invention will be better understood with reference to the detailed description when considered in conjunction with the non-limiting examples and the accompanying drawings, in which:

[0027] [Fig. 1] shows a diagram illustrating a communication arrangement for usage of an e-hailing service, including a smartphone and a server; [0028] [Fig. 2] shows a diagram of an architecture for task allocation in parallel streams based on a dynamic resource state, according to embodiments;

[0029] [Fig. 3] shows a flow diagram illustrating a method for task allocation in parallel streams based on a dynamic resource state, according to embodiments;

[0030] [Figs. 4A-4I] show diagrams of an example of task allocation in parallel streams based on a dynamic resource state, according to embodiments; and

[0031] [Fig. 5] shows a block diagram of the server of [Fig. 1], implementing the architecture of [Fig. 2],

DETAILED DESCRIPTION

[0032] The following detailed description refers to the accompanying drawings that show, by way of illustration, specific details and embodiments in which the disclosure may be practiced. These embodiments are described in sufficient detail to enable those skilled in the art to practice the disclosure. Other embodiments may be utilized and structural, and logical changes may be made without departing from the scope of the disclosure. The various embodiments are not necessarily mutually exclusive, as some embodiments can be combined with one or more other embodiments to form new embodiments.

[0033] Embodiments described in the context of one of the devices or methods are analogously valid for the other devices or methods. Similarly, embodiments described in the context of a device are analogously valid for a vehicle or a method, and vice-versa.

[0034] Features that are described in the context of an embodiment may correspondingly be applicable to the same or similar features in the other embodiments. Features that are described in the context of an embodiment may correspondingly be applicable to the other embodiments, even if not explicitly described in these other embodiments. Furthermore, additions and/or combinations and/or alternatives as described for a feature in the context of an embodiment may correspondingly be applicable to the same or similar feature in the other embodiments. [0035] In the context of various embodiments, the articles “a”, “an” and “the” as used with regard to a feature or element include a reference to one or more of the features or elements.

[0036] As used herein, the term “and/or” includes any and all combinations of one or more of the associated listed items.

[0037] In the following, embodiments will be described in detail.

[0038] An e-hailing app, typically used on a smartphone, allows its user to hail a taxi or also a private driver through his or her smartphone for a trip.

[0039] [Fig. 1] shows a diagram of a communication arrangement 100 for usage of an e-hailing service, including a smartphone 105 and a server 115 (computer).

[0040] The smartphone 105 has a screen showing a graphical user interface (GUI) 106 of an e-hailing app that a user of the smartphone 105 previously installed on his smartphone and opened (i.e. , started) to e-hail a ride (taxi or private driver).

[0041] The GUI 106 includes a map 107 of a vicinity of a position of the user, which the app may determine based on a location service, e.g., a GPS-based location service. Further, the GUI 106 includes a box for a point of departure 108, which may be set to the user’s current location obtained from the location service, and a box for a destination 109, which the user may touch to enter the destination, e.g., opening a list of possible destinations. There may also be a menu (not shown) allowing the user to select various options, e.g., how to pay (cash, credit card, credit balance of the e-hailing service). When the user selects the destination and makes any necessary option selections, he or she may touch a “find car” button 110 to initiate searching of a suitable car.

[0042] For the above, the e-hailing app communicates with the server 115 of the e-hailing service via a radio connection. The server 115 may consult a memory of the server 115 or a data storage 121 having information of current locations of registered vehicles 111 , about when they are expected to be free, about traffic jams, etc. From this, a processor of the server 115 selects the most suitable vehicle (if available, i.e., if a request can be fulfilled) and provides an estimate of time when a driver will be there to pick up the user, a price of a ride and how long it will take to get to the destination. The server 115 communicates this back to the smartphone 105, and the smartphone 105 displays this information on the GUI 106. The user may then accept (i.e. , book) by touching a corresponding button. If the user accepts, the server 115 informs a selected one among the vehicles 111 (or, equivalently, its driver), i.e., the vehicle the server 115 has allocated for fulfilling the transport request.

[0043] It should be noted while the server 115 is described as a single server, its functionality, e.g., for providing an e-hailing service for a whole city, will in practical application typically be provided by an arrangement of multiple server computers (e.g., implementing a cloud service). Accordingly, functionalities described in the following provided by the server 115 may be understood to be provided by an arrangement of servers or server computers.

[0044] The data storage 121 may, for example, be part of a cloud-based system 120 provided by a cloud storage provider to store and access data, which it may use for taking decisions, such as information of locations of passengers and vehicles, their history (earlier bookings and routes taken), etc.

[0045] The server 115 together with the vehicles 111 provide the e-hailing service, i.e., forms a transport system. It should be noted that while the example of [Fig. 1] relates to an e-hailing service where persons are transported, a transport system providing a transport service for transporting other items like fresh food and parcels may similarly be provided.

[0046] When a user makes a booking for a transport task, the server 115 may provide the smartphone 105 with an estimate of time when the transport task is completed, e.g., when the user will arrive, when food will be delivered to the user, etc.

[0047] The server 115 may receive streams of an e-hailing task and a transport task at the same time or in parallel, both of these tasks including computation or arithmetic. When this occurs, the server 115 can allocate the e-hailing task and the transport task in the parallel streams, to its CPU and memory or its computational resources, by consulting a dynamic resource state stored in the memory of the server 115 or the data storage 121. In detail, the allocation of the computational tasks in the parallel streams may be based on consideration of a computational capacity of each node in a cluster of the computational resources. [0048] The embodiments described herein may make full use of the free computational resources in the cluster, while ensuring load balancing of services and reducing a risk of downtime. The embodiments can also address a problem of not considering an adaptation of job and node resources and service load balancing when parallelizing execution of tasks. Further, the embodiments may reduce a service failure rate and improve resource utilization.

[0049] [Fig. 2] shows a diagram of an architecture 200 for task allocation in parallel streams based on a dynamic resource state 230, according to embodiments.

[0050] Referring to [Fig. 2], the architecture 200 may include a stream application program interface (API) service portion 210, a pipeline service portion 215, a job manager 220, a resource manager 225, the dynamic resource state 230 and task managers 235, 240, 245 and 250.

[0051] The stream API service portion 210 receives parallel streams of tasks from a collection of service applications 205 (e.g., an e-hailing service application “GrabExpress” and a transport service application “GrabTransport”) installed on a smartphone, e.g., the smartphone 105 of [Fig. 1], Thus, the embodiments described herein may be applied to all businesses of an e-service company using parallel streams handling. The stream API service portion 210 parses the received parallel streams into data flows.

[0052] The pipeline service portion 215 pipelines the data flows into which the streams are parsed, into a sequence of jobs (a chain of operations). A job may be a single arithmetic operator or a set of multiple arithmetic operators. The pipeline service portion 215 outputs each individual one of the jobs.

[0053] The job manager 220 assigns each individual one of the jobs, which is not split into new ones, to a sequential place in a job queue included in the job manager 220. Data about the multiple jobs are already in their own tasks, are unrelated to each other, and may be executed independently. The job queue holds the incoming jobs on a first-in-first-out (FIFO) basis. The jobs may be inserted into a tail of the job queue.

[0054] The job manager 220 further estimates a number of logical CPUs (e.g., cores) and a size of logical memory (in gigabytes (GB)) for executing each individual job. The number of logical CPUs on a single node among multiple nodes in a cluster of computational resources can equal a number of physical CPUs multiplied by a number of cores per physical CPU multiplied by a number of hyperthreads.

[0055] The resource manager 225 may obtain each of the assigned jobs from a head of the job queue included in the job manager 220, and allocates each of the obtained jobs to a node or one of the task managers 235, 240, 245 and 250, based on the dynamic resource state 230, to adapt the computational resources of the cluster. Each of the obtained jobs may be allocated further based on allocation policies further described with respect to [Fig. 3] below. The dynamic resource state 230 dynamically (in real-time) keeps track of, for each of the nodes, a number of idle logical CPUs, a size of idle logical memory in GB, and a number of tasks executing at a respective one of the nodes. In detail, the resource manager 225 scans and reads the dynamic resource state 230 to locate the appropriate node for each of the jobs, based on the allocation policies, and allocates a respective one of the jobs to the located appropriate node.

[0056] Each of the task managers 235, 240, 245 and 250 is located on a respective node in the cluster. Each of the task managers 235, 240, 245 and 250 executes the job allocated to a respective one of the task managers 235, 240, 245 and 250. Each of the task managers 235, 240, 245 and 250 further reports, to the resource manger 225, the number of idle logical CPUs and the size of idle logical memory (computational resources) of the respective one of the task managers 235, 240, 245 and 250. When a new job arrives, one of the task managers 235, 240, 245 and 250 generates a new task using a number of logical CPUs and a size of logical memory for the new job, and increments a number of tasks executing at the one of the task managers 235, 240, 245 and 250. When the one of the task managers 235, 240, 245 and 250 finishes executing the new task, the one of the task managers 235, 240, 245 and 250 releases the computational resources used by the new task, and reports, to the resource manager 225, a resource status of the released computational resources and the number of tasks that is decremented.

[0057] [Fig. 3] shows a flow diagram illustrating a method 300 for task allocation in parallel streams based on the dynamic resource state 230, according to embodiments. [0058] In operation 305, the method 300 includes the resource manager 225 or the job manager 220 obtaining a job from jobs to be executed for service applications, and inserting the job to the tail of the job queue.

[0059] In operation 310, the method 300 includes the resource manager 225 reading a job from the head of the job queue. The job entry and exit queue follows the FIFO rule.

[0060] In operation 315, the method 300 includes the resource manager 225 scanning the dynamic resource state 230 to locate a matching node in the cluster for the read job, based on a standard of the matching node meeting or having available a number of logical CPUs and a size of logical memory for executing the job.

[0061] In operation 320, the method 300 includes the resource manager 225 determining whether the matching node exists. Based on the matching node being determined to exist, the method 300 continues in operation 325. Otherwise, the method 300 continues in operation 330.

[0062] In operation 325, the method 300 includes the resource manager 225 determining whether multiple matching nodes exists. Based on the multiple matching nodes being determined to exist, the method 300 continues in operation 335. Otherwise the method 300 continues in operation 340.

[0063] In operation 330, the method 300 includes the resource manager 225 waiting for an update signal to be received before returning to operation 310. The update signal indicates that the dynamic resource state 230 is updated.

[0064] In operation 335, the method 300 includes the resource manager 225 selecting the matching node determined to exist.

[0065] In operation 340, the method 300 includes the resource manager 225 selecting, from the multiple matching nodes, one or more nodes having a least number of tasks executing therein. This balances service load on all of the nodes in the cluster.

[0066] In operation 345, the method 300 includes the resource manager 225 determining whether multiple nodes having the least number of tasks executing therein are selected. Based on the multiple nodes having the least number of tasks executing therein being determined to be selected, the method 300 continues in operation 350. Otherwise the method 300 continues in operation 355. [0067] In operation 350, the method 300 includes the resource manager 225 selecting, from the multiple nodes having the least number of tasks executing therein, a node having a lowest similarity value between the read job a respective one of the nodes. The similarity value may be a difference value between the number of logical CPUs for executing the read job and a number of logical CPUs available in the respective node, added by a difference value between the size of logical memory for executing the read job and a number of logical memory available in the respective node. The lower the similarity value, the more similar the computational resources for executing the job is to the computational resources in the respective one of the nodes. The resource manager 225 may calculate a similarity value S between the read job and each of the nodes, based on the dynamic resource state 230 and the following equation: when N _lcpu > R _lcpu and

[0069] N represents a number or size of computational resources available in a respective node, R represents a number or size of computational resources for executing a job, Icpu represents a number of logical CPUs, and memory represents a size of logical memory.

[0070] In operation 355, the method 300 includes the resource manager 225 updating the dynamic resource state 230 to account for computational resources (e.g., a number of logical CPUs and a size of logical memory) of the selected node being used. The method 300 further includes the resource manager 225 generating the update signal indicating that the dynamic resource state 230 is updated and that a job may be read from the job queue included in the job manager 220.

[0071] In operation 360, the method 300 includes the resource manager 225 allocating the job out from the job queue to the selected node.

[0072] In operation 365, the method 300 includes the task manager 235, 240, 245 or 250 on the selected node allocating the computational resources to the allocated job.

[0073] In operation 370, the method 300 includes the task manager 235, 240, 245 or 250 generated a new task for the allocated job, based on the computational resources allocated to the job. [0074] In operation 375, the method 300 includes the task manager 235, 240, 245 or 250 executing the generated new task.

[0075] In operation 380, the method 300 includes the task manager 235, 240, 245 or 250 determining whether the executed new task is complete. Based on the executed new task being determined to be complete, the method 300 continues in operation 385. Otherwise, the method 300 returns to operation 375.

[0076] In operation 385, the method 300 includes the task manager 235, 240, 245 or 250 releasing the computational resources allocated to the job, and reporting the free released computational resources to the resource manager 225.

[0077] In operation 390, the method 300 includes the resource manager 225 updating the dynamic resource state 230 to account for the computational resources (e.g., the number of logical CPUs and the size of logical memory) of the selected node being released. The method 300 further includes the resource manager 225 generating the update signal indicating that the dynamic resource state 230 is updated and that a job may be read from the job queue included in the job manager 220.

[0078] The method 300 of [Fig. 3] is, for example, carried out by the server 115 as illustrated in [Fig. 5],

[0079] [Figs. 4A-4I] show diagrams of an example 400 of task allocation in parallel streams based on the dynamic resource state 230, according to embodiments.

[0080] Referring to [Figs. 4A and 4B], there are four nodes in a cluster of computational resources, and idle computational resources in the dynamic resource state 230 are at initialization.

[0081] Referring to [Figs. 4A and 4C], JobO in the job queue included in the job manager 220 uses 1 logical CPU and 1 GB of logical memory. The resource manager 225 scans the dynamic resource state 230 and determines that all of the nodes match or include enough available computational resources to be used to execute JobO. The resource manager 225 then calculates a similarity value between JobO and each of the nodes, and selects Node2 corresponding to the task manager 245 that has a lowest similarity value of 1 with JobO. The resource manager 225 allocates JobO to selected Node2, namely, a slot of the task manager 245, as shaded. The resource manager 225 updates the dynamic resource state 230 by decrementing computational resources allocated to JobO from those of Node2 and incrementing a number of tasks of Node2 by 1 , as shaded.

[0082] Referring to [Figs. 4A and 4D], Job1 in the job queue also uses 1 logical CPU and 1 GB of logical memory. The resource manager 225 scans the dynamic resource state 230 and determines that NodeO, Nodel and Node3 match or include enough available computational resources to be used to execute Job1. The resource manager 225 then calculates a similarity value between Job1 and each of NodeO, Nodel and Node3, and selects NodeO corresponding to the task manager 235 that has a lowest similarity value of 3 with Job1. The resource manager 225 allocates Job1 to selected NodeO, namely, a slot of the task manager 235, as shaded. The resource manager 225 updates the dynamic resource state 230 by decrementing computational resources allocated to Job1 from those of NodeO and incrementing a number of tasks of NodeO by 1 , as shaded.

[0083] Referring to [Figs. 4A and 4E], Job2 in the job queue uses 2 logical CPUs and 10 GB of logical memory. The resource manager 225 scans the dynamic resource state 230 and determines that only Node3 matches or includes enough available computational resources to be used to execute Job2. The resource manager 225 allocates Job2 to Node3, namely, a slot of the task manager 250, as shaded. The resource manager 225 updates the dynamic resource state 230 by decrementing computational resources allocated to Job2 from those of Node3 and incrementing a number of tasks of Node3 by 1 , as shaded.

[0084] Referring to [Figs. 4A and 4F], Job3 in the job queue uses 1 logical CPU and 2 GB of logical memory. The resource manager 225 scans the dynamic resource state 230 and determines that Nodel and Node3 match or include enough available computational resources to be used to execute Job3. The resource manager 225 selects Nodel corresponding to the task manager 240 that has a least number of tasks executing therein. The resource manager 225 allocates Job3 to Nodel , namely, a slot of the task manager 240, as shaded. The resource manager 225 updates the dynamic resource state 230 by decrementing computational resources allocated to Job3 from those of Nodel and incrementing a number of tasks of Nodel by 1 , as shaded. [0085] Referring to [Figs. 4A, 4G, 4H and 4I], Job4 in the job queue uses 2 logical CPUs and 8 GB of logical memory. In [Fig. 4G], the resource manager 225 scans the dynamic resource state 230 and determines that none of the nodes matches or includes enough available computational resources to be used to execute Job4. In [Fig. 4H], the resource manager 225 waits until the execution of Job2 is completed and the computational resources allocated to Job2 are released and incremented back to those of Node3, as emphasized. In [Fig. 4I], the resource manager 225 selects Node3 corresponding to the task manager 250 that is the only node that matches or includes enough available computational resources to be used to execute Job4. The resource manager 225 allocates Job4 to Node3, namely, a slot of the task manager 250, as shaded. The resource manager 225 updates the dynamic resource state 230 by decrementing computational resources allocated to Job4 from those of Node3 and incrementing the number of tasks of Node3 by 1 , as shaded.

[0086] [Fig. 5] shows a block diagram of the server 115 of [Fig. 1], implementing the architecture 200 of [Fig. 2],

[0087] Referring to [Fig. 5], the server 115 may be a server computer that includes a communication interface 505, a processor 510 and a memory 515.

[0088] The communication interface 505 may serve as a hardware and/or software interface that can, for example, transfer commands and/or data between a user and/or external devices and other components of the server 115. The communication interface 505 may further set up communication between the server 115 and the external devices, such as the smartphone 105 of [Fig. 1], The communication interface 505 may be connected with a network through wireless or wired communication architecture to communicate with the external devices. The communication interface 505 may be a wired or wireless transceiver or any other component for transmitting and receiving signals.

[0089] The processor 510 may include one or more of a CPU, a graphics processor unit (GPU), an accelerated processing unit (APU), a many integrated core (MIC), a field-programmable gate array (FPGA), and/or a digital signal processor (DSP). The processor 510 may be a general-purpose controller that performs control of any one or any combination of the other components of the server 115, and/or performs an operation or data processing relating to communication. The processor 510 may execute one or more programs stored in the memory 515.

[0090] The memory 515 may include a volatile and/or non-volatile memory. The memory 515 stores information, such as one or more of commands, data, programs (one or more instructions), applications, etc., which are related to at least one other component of the server 115 and for driving and controlling the server 115. For example, commands and/or data may formulate an operating system (OS). Information stored in the memory 515 may be executed by the processor 510. The memory 515 may further store information that is executed by the processor 510 to perform functions and operations described with respect to [Figs. 1-41] above.

[0091] The methods described herein may be performed and the various processing or computation units and the devices and computing entities described herein may be implemented by one or more circuits. In an embodiment, a "circuit" may be understood as any kind of a logic implementing entity, which may be hardware, software, firmware, or any combination thereof. Thus, in an embodiment, a "circuit" may be a hard-wired logic circuit or a programmable logic circuit such as a programmable processor, e.g., a microprocessor. A "circuit" may also be software being implemented or executed by a processor, e.g., any kind of computer program, e.g., a computer program using a virtual machine code. Any other kind of implementation of the respective functions that are described herein may also be understood as a "circuit" in accordance with an alternative embodiment.

[0092] While the disclosure has been particularly shown and described with reference to specific embodiments, it should be understood by those skilled in the art that various changes in form and detail may be made therein without departing from the spirit and scope of the invention as defined by the appended claims. The scope of the invention is thus indicated by the appended claims and all changes that come within the meaning and range of equivalency of the claims are therefore intended to be embraced.