Configuration of asynchronous message processing in dataflow networks

Managing a concurrency based system. A method includes determining a defined number of concurrent operations for a concurrency based management system. An operation queue is accessed. The operation queue stores or has therein zero or more asynchronous operations. An asynchronous operation is an operation that returns an object representing the asynchronous completion of the operation. The method further includes, as long as the queue is not empty, scheduling asynchronous operations from the queue until a number of asynchronous operations equal to the defined number has been reached.

BACKGROUND

Background and Relevant Art

A dataflow network is a network of concurrently executing processes or automata referred to as nodes that can communicate with each other by virtue of tree or graph edges connecting the nodes. Dataflow nodes in a dataflow network exist to process messages provided to them as input and/or output messages to other target nodes or receiving code.

Data operations may be of a synchronous type or asynchronous type. A synchronous operation is an operation that can block the caller on a user function. A processor or thread begins processing the operation, but requires input or processing by others before the operation can be completed. The processor or thread waits idle for the input or processing and completes processing the operation once the input or processing by others is received or completed. In summary, for synchronous operations, the operation's completion signals the end of the message's processing.

An asynchronous operation does not block the caller on a user function. A processor or thread begins processing an operation, but requires input or processing by others before finishing processing the operation. Rather than waiting idle for the input or processing, the processor or thread can perform other function until the input or processing by others is completed or received, and then returns to finish processing the operation once the input or processing by others is received or completed. In other words, for asynchronous operations, a first operation may asynchronously signal to the system upon the operation's completion for the first operation to be considered completed.

While dataflow systems provide support for dataflow blocks, these dataflow blocks may support processing only one message at a time, and they are limited to synchronous processing, meaning that the processing of a message is considered to start when a function to process that message is invoked and to complete when that function exits. It would be useful for dataflow network systems to be able to manage synchronous and asynchronous operations and to support multiple messages processing.

In dataflow systems, there is some overhead when assigning operations to processing blocks. In some cases, the overhead of assigning an operation may actually be more expensive in terms of computing resources to schedule than it is to simply perform the operation. Thus, it would be useful to reduce the amount of scheduling needing in dataflow processing systems.

BRIEF SUMMARY

One embodiment illustrated herein includes a method practiced in a computing environment. The method includes acts for managing a concurrency based system. The method includes determining a defined number of concurrent operations for a concurrency based management system. An operation queue is accessed. The operation queue stores or has therein zero or more asynchronous operation messages indicating operations to be performed. An asynchronous operation is an operation is that returns an object representing the asynchronous completion of the operation. The method further includes, so long as the queue is not empty, scheduling asynchronous operations from the queue until a number of asynchronous operations equal to the defined number has been reached.

DETAILED DESCRIPTION

Some example embodiments described herein implement a concurrency based management system whereby synchronous and asynchronous operations can be scheduled and performed by a dataflow system. In particular, synchronous and asynchronous operations can be performed in parallel based on concurrency capabilities of a dataflow block.

As noted, some dataflow systems provide support for dataflow blocks. These dataflow blocks support processing only one message at a time, and they are limited to synchronous processing, meaning that the processing of a message is considered to start when a function to process that message is invoked and to complete when that function exits. However, some embodiments herein support multiple messages of both synchronous and asynchronous types, based on concurrency capabilities.

In particular, dataflow processing blocks maintain a count for the number of outstanding operations, which are represented herein as tasks. These tasks may be represented as “futures” (as illustrated in more detail below) in the case of asynchronous operations. Synchronous operations can be converted to asynchronous operations by processing a quantity of the synchronous operations. A quantity of synchronous operations can be represented by a future as well.

When new messages arrive, additional processing futures are spun up if the number of outstanding operations is below the maximum threshold of the concurrency capabilities. These processing futures launch asynchronous operations, each of which is tracked individually, and the count is managed based on when the launched futures complete. Upon completion of a future, if more messages are available to be processed, further futures may be launched to process those messages, again subject to the maximum concurrency threshold. Alternatively, embodiments may be implemented that are able to reuse already launched futures to process messages. This can result in an overhead savings, as a new future will not need to be spun up to process the message.

Referring now toFIG. 1, an example is illustrated.FIG. 1illustrates a simple dataflow system100. The dataflow system100includes a data source block102and a target block104. The data source block102sends messages (referred to generally as106, but shown inFIG. 1specifically as106-1,106-2, and106-n) to the target block104for processing. The target block can accept and process multiple messages106. While in this simple example, messages106are sent from the data source block102to a target block104, it should be appreciated that the dataflow network may include many blocks, each of which performs some function on data provided to it from a source block. For example, the target block104may be a source block for another block downstream from the target block104.

Referring now toFIG. 2A, additional details of the target block104are illustrated. The target block104has concurrency capabilities in that in can process multiple messages at any given time. In the example illustrated, the target block104has a concurrency capability of five. Additionally, the target block is executing at its maximum concurrency of five as illustrated by the five futures, referred to herein generally as108, but shown specifically as108-1,108-2,108-3,108-4, and108-5. While the target block is shown executing at its maximum capacity of five futures108, it should be appreciated that the target block104may be running any number of futures from zero to five futures. Additionally, other systems may have different concurrency capabilities and the example of a maximum concurrency threshold of five is only illustrative. Further, while only a single block104is illustrated, it should be appreciated that different systems can have more blocks which also have separate concurrency capabilities. Each of the different futures108in the different blocks of a dataflow system draws upon a thread set to access hardware computing resources for performing the futures108.

FIG. 2Afurther illustrates a thread set112. The thread set represents the threads, referred to generally as114, but illustrated specifically as114-1,114-2, and114-3, which can be used to perform the futures108. Notably, the thread set112does not need to include a number of threads114equal to the concurrency threshold of one or all blocks in the dataflow system100. For example, concurrency thresholds can be higher by virtue of thread sharing or freeing up of threads due to asynchronous futures being performed. For example, an example asynchronous future may be a download future. The asynchronous download future108may include requesting that some data be downloaded from an external source. A thread114is used by the asynchronous future108to make the download request and is then freed to be used by another future while the data downloads. Once the data has downloaded, the asynchronous future may request use of a thread114to perform further processing.

The threads114illustrated herein may take any one of a number of different forms. For example, each thread may be the single thread included as part of a single processor core in a multi-core system. In particular, a processor core may have a single arithmetic logic unit (ALU) and a single set of thread registers which are comprised of the single thread. Alternatively, each thread may be one thread from among a plurality of threads of a single processor. For example, a core may have a single ALU, but multiple sets of thread registers, such that multiple threads are implemented using a single core. Alternatively, virtual threads may be implemented where an operating system or other software manages virtual threads and schedules the virtual thread to execute on underlying hardware processors or hardware threads.

Illustrating now more detailed examples of functionality, in some embodiments, a dataflow processing block, e.g. target block104, maintains a count of the number of messages that may be processed concurrently. In the example shown inFIGS. 1 and 2, that number is five. When new messages106arrive into the target block's input queue110, the target block104checks whether the current concurrency level is less than the maximum level allowed, i.e. five. If it is, new futures108may be spun up to execute a user-provided function for the input message106, up to the maximum concurrency level allowed. That user-provided function may be deemed synchronous or asynchronous. For synchronous, the function's completion signals the end of the message's processing. The future exists and a number tracking the current number of futures108for the block104, i.e. the concurrency level, is decremented. For asynchronous functions, the function returns a future to represent the asynchronous completion of the message's processing. That future's108completion signals the end of the message's processing. When a message's processing is completed, the count of the current concurrency level may be decremented, allowing other futures108to be spun up. As will be illustrated below, other embodiments may reuse futures to reduce the amount of overhead needed for message processing. In particular, a single future may be used to processes a quantity of similar or identical messages.

As discussed above, processing of asynchronous operations are represented using a future type. The future type may support multiple completion states: RanToCompletion, Faulted, and Canceled. The first two states are direct corollaries to a synchronous invocation of a function either running to completion or causing an error such as throwing an exception out of the function. The third state may potentially be triggered by input, such as a specific kind of exception that indicates the operation was canceled, or it may be achieved through directly information the “future” that the operation was canceled. Embodiments may be implemented where when the processing of an asynchronous operation completes, the completion status, final result, and exception information is all available from the future representing the operation. Continuation operations may be registered with the future object to be notified of its completion, and to perform operations such as decrementing the concurrency count mentioned previously.

When new messages arrive at a target block104and the maximum concurrency level has been reached, further data may be buffered in input queues, such as queue110, until such time that resources are available to process those messages.

When futures108complete the processing of a message106, they may attempt to minimize overheads by picking off the next message from the input queues110and processing it, rather than returning completely and forcing a new future to be spun up to process the next message in the queue110. However, to aid in fairness of processing across multiple dataflow blocks, this behavior may be configurable in a manner that enables control of the maximum number of messages (illustrated as a quantity below) an individual future108may process. In one embodiment, the future itself may be defined as to the number of messages106that it can process before the future ends. This allows other futures108to have an opportunity to be spun up and executed. If a future108processing messages106finds that it has already processed the maximum number (i.e. the quantity), it will retire itself, in the process potentially spinning up a replica of itself to continue the processing. This allows the underlying thread114that was running the future108to be released and to process other futures waiting for processing time.

Illustrating now a specific example, a first set of messages may be sent to the target block104. The first set of messages may include 100 messages to be processed by a given function. The target block may be configured to create functions that only allow a future108to process 10 messages before the future108is retired. A different set of messages may also be sent to the target block104. The target block may spin up a future108to process messages from the first set of messages. The future108will process a message from the set of 100 messages and once the message is processed will return to the queue110to obtain the next message from the set of 100 messages. This will continue until the first ten messages of the set of 100 messages have been processed. The future108will then retire itself releasing a thread114being used to perform the future108. This allows the thread to be used by a different future that is then spun up to process one or more of the messages in the different set of messages. The future108, when a thread is available, will spin up a replica of itself to process an additional 10 messages from the set of 100 messages.

These futures may also be scheduled to run on custom scheduling environments. A developer provides as a configuration option to the dataflow block a scheduler entity116which controls how the generated futures are executed. The dataflow target block110creates its futures and hands them off to the scheduler, which then decides where and when to execute the futures. Further, the functions in these futures execute in an environment where the provided scheduler116is exposed as TaskScheduler.Current, meaning that any futures generated during the processing of that message will also be scheduled to that scheduler. This enables asynchronous functions to easily break up their processing into multiple individual futures, each of which will automatically by default be scheduled to the target scheduler116.

Referring now toFIG. 2B, an alternative embodiment for scheduler components is illustrated.FIG. 2Billustrates that the target block104includes a pair of schedulers116aand116b. In the example illustrated there is a pair of schedulers that collude so that each scheduler colluding knows what the other scheduler is doing, including a concurrent scheduler116aand an exclusive scheduler116b. In the case of the exclusive scheduler116a, any future scheduled to the exclusive scheduler116awill be guaranteed to be an exclusive future being the only future that is running at a given time on either scheduler116aor116bin the target block104. The concurrent scheduler will be able to run any number of futures concurrently as long as there are not any exclusive futures running. For example, this scheduler arrangement may be used to implement a reader/writer scenario where any number of readers can be executing as long as they are only readers and not writers. But only one writer future is executing at a time as long as there is no other read or write futures executing. Embodiments could have action blocks configured to be on the exclusive scheduler116a. In this way all the work that the blocks schedule will in effect be serialized to eliminate threading issues.

Some embodiments may be implemented using functionality in .NET 4 framework available from Microsoft Corporation of Redmond Wash. For example, dataflow processing blocks maintain a count for the number of outstanding operations, all of which are represented as “futures.” In .NET 4, this may be accomplished using System.Threading.Task.Task and System.Threading.Task.Task<TResult>. When new messages arrive, additional processing futures are spun up if the number of outstanding operations is below the maximum threshold. These processing futures launch asynchronous operations, each of which is tracked individually, and the count is managed based on when the launched futures complete; upon their completion, if more messages are available to be processed, further futures may be launched to process those messages, again subject to the maximum threshold.

A dataflow processing block maintains a count of the number of messages that may be processed concurrently. When new messages arrive into the block's input queue, the block checks whether the current concurrency level is less than the maximum level allowed. If it is, new futures may be spun up to execute a user-provide function for the input message, up to the maximum concurrency level allowed. That user-provided function may be deemed synchronous or asynchronous. For synchronous, the function's completion signals the end of the message's processing; for asynchronous, the function returns a future, i.e. a Task or a Task<TResult> in .NET 4, to represent the asynchronous completion of the message's processing. That future's completion signals the end of the message's processing. When a message's processing is completed, the count may be decremented.

Processing of asynchronous operations is represented using a future type that supports multiple completion states: RanToCompletion, Faulted, and Canceled. The first two states are direct corollaries to a synchronous invocation of a function either running to completion or throwing an exception out of the function; the third state is may potentially be triggered by a specific kind of exception that indicates the operation was canceled, or it may be achieved through directly information the “future” that the operation was canceled (e.g. SetCanceled( )). In the .NET solution, this is handled by the .NET Task and Task<TResult> type. When the processing of an asynchronous operation completes, the completion status, final result (in the case of a Task<TResult>), and exception information is all available from the future representing the operation.

Referring now toFIG. 3, a method300is illustrated. The method300may be practiced in a computing environment and includes acts for managing a concurrency based system. The method300includes determining a defined number of concurrent operations for a concurrency based management system (act302). For example, inFIG. 2A, the system has a concurrency limit of five.

The method300further includes accessing an operation queue (act304). The operation queue stores or has therein zero or more asynchronous operation request messages. An asynchronous operation is an operation that returns an object representing the asynchronous completion of the operation. For example, the queue110may have a number messages stored therein.

The method300further includes, so long as the queue is not empty, scheduling asynchronous operations from the queue until a number of asynchronous operations equal to the defined number has been reached (act306). For example, embodiments may schedule five asynchronous operations using the scheduler116from the queue110for executions as operations unless the queue is empty before five asynchronous operations can be scheduled.

The method300may be practiced where scheduling operations from the queue includes converting synchronous operations to asynchronous operations. A synchronous operation is an operation in which the operation's completion signals the end of a message's processing.

In some embodiments, converting synchronous operations to asynchronous operations includes reusing the scheduling act to process a quantity of synchronous operations as a set. For example, a quantity of synchronous operations that have the same future performed for them can be scheduled to be executed as a group converting the group of synchronous operations to a single asynchronous operation. In some embodiments, the quantity is limited to a defined number of synchronous operations. For example, as illustrated above, a first set of messages may include 100 messages to be processed by a given function. The target block may be configured to create functions that only allow a future108to process 10 messages before the future108is retired. If additional messages from the first set of messages remain, an additional process to process 10 messages can be spun up. However, by limiting the number of messages that can be processed in a quantity, threads will not be blocked by a single quantity for an undue extended period of time. In some embodiments, the defined number of synchronous operations is dynamic number. In alternative embodiments, the defined number of synchronous operations is a static pre-determined number.

Further, upon reaching various computer system components, program code means in the form of computer-executable instructions or data structures can be transferred automatically from transmission computer readable media to physical computer readable storage media (or vice versa). For example, computer-executable instructions or data structures received over a network or data link can be buffered in RAM within a network interface module (e.g., a “NIC”), and then eventually transferred to computer system RAM and/or to less volatile computer readable physical storage media at a computer system. Thus, computer readable physical storage media can be included in computer system components that also (or even primarily) utilize transmission media.