System and method for synchronizing message processing in a continuous processing system

This application claims the benefit of U.S. Provisional Application No. 60/819,302 filed on Jul. 7, 2006 with inventors Aleksey Sanin, Ian D. Marshall, and Giuliano Carlini and titled “DB Joiner and Passive Synchronizers: Real Query Processor ‘Timezones,’” the contents of which are incorporated by reference as if fully disclosed herein.

1. Field of the Invention

This invention relates generally to continuous processing systems that process streaming data, and, more specifically, to synchronizing message processing in a continuous processing system.

2. Description of the Background Art

A continuous processing system processes streaming data. It includes statements (such as queries), written by programmers, which operate continuously on input data streams and which publish to output data streams. In such system, it is often difficult to achieve predictable and repeatable output results.

When statements written by programmers are compiled, an execution graph may be created, where the execution graph is comprised of connected primitives that correspond to the compiled statements. An execution graph in a continuous processing system specifies the path for processing messages in accordance with the statements. In other words, the continuous processing system processes messages by pushing them through the execution graph.

In order to achieve predicable and repeatable output results, messages have to be processed in accordance with message order rules (i.e., rules that specify the order in which messages need to be processed). For example, in one embodiment, messages are assigned an internal timestamp and are processed in order of their timestamp, where messages with the same timestamp are processed together.

Certain types of primitives in an execution graph may have potential for substantial delay. Examples of such primitives include primitives that make database calls or remote procedure calls, as well as primitives associated with user-defined functions.

For efficiency and speed purposes, it is often desirable to process messages in parallel, which means that more than one row from a data stream may enter the execution graph at a particular time. With parallel processing, a primitive may process multiple messages with different timestamps at the same time. For example, a primitive that joins data in a message with data in a database (a “database joiner primitive” or “DB Joiner”) may make concurrent database calls for multiple joins at once.

If parallel processing occurs in a graph that has primitives with potential for substantial delay, messages can easily get processed out of order, which means that the output results will not be predictable and repeatable. FIG. 1 illustrates an example of this. In this example, messages from data stream 110 go into both database joiner 120 and joiner 130 . Message order rules for this example system require that a message with timestamp x be joined at joiner 130 with output of database joiner 120 for a message with timestamp x (i.e., a message with the same timestamp). If only one message (i.e., one row) from data stream 110 goes into the execution graph at a time, the message order is easily preserved. However, if multiple messages enter the graph at once, messages could be processed out of order if one primitive is slower than the other primitive. In this example, messages with timestamps 1 - 5 go into the graph 100 and the database joiner 120 is delayed, thereby causing the output of the database joiner 120 for message 1 to be joined with message 5 in stream 110 instead of message 1 (where the correct result would be to join the output of the database joiner for message 1 with message 1 ).

Therefore, in an execution graph where there are primitives with the potential for substantial delay, there is a need for a system and method that permits parallel processing within some areas of the execution graph while protecting other areas of the graph from processing messages out of order due to primitives with potential for substantial delay.

In cases where an execution graph in a continuous processing system includes primitives with potential for substantial delay, the present invention provides a system and method for synchronizing the processing of messages in areas of the execution graph affected by (i.e., dependent on) primitives with potential for substantial delay. The invention is especially useful in cases where it is desirable to be able to process messages with different internal timestamps in parallel (i.e., concurrently) in certain areas of the execution graph while maintaining serial processing in other areas. For instance, it enables a database joiner primitive to make concurrent calls to a database while still maintaining the proper order in which messages downstream of the database joiner should be processed.

In the present invention, the continuous processing system identifies primitives with potential for substantial delay. Primitives with potential for substantial delay can cause messages to be processed in an incorrect order when messages with different timestamps are in the execution graph at the same time.

For each potentially-delayed primitive that is not in a loop, the continuous processing system creates a “timezone” downstream of such primitive. A timezone is an area of the execution graph that, at any given time, includes only messages that need to be processed together (e.g., messages with the same internal timestamp). All primitives downstream of the potentially-delayed primitive are considered part of the timezone. The output(s) of the potentially-delayed primitive is an input to the timezone.

The continuous processing system then identifies all other inputs to the timezone and places a synchronizer across all inputs to the timezone. The synchronizer controls the flow of messages into the timezone to ensure that messages that need to be processed together enter the timezone at the same time. The synchronizer does not let additional messages into the timezone until the previously-entered messages have been processed by the timezone.

If a primitive with potential for substantial delay is part of a loop, then such primitive itself becomes part of the timezone, and a synchronizer is placed across all inputs to the timezone, including any feedback paths to the potentially-delayed primitive.

The use of timezones allows certain areas of the execution graph (i.e., areas outside the timezone) to process messages with different timestamps in parallel, while protecting other areas of the graph from processing messages out of order due to primitives with potential for substantial delay. Parallel processing of messages with different internal timestamps does not occur within a timezone.

FIG. 1 is a block diagram that illustrates an example execution graph with parallel message processing.

FIG. 2 is a block diagram that illustrates one embodiment of a continuous processing system.

FIG. 3 is a flow chart that illustrates a method for synchronizing the processing of messages in areas of the execution graph affected by primitives with potential for substantial delay.

FIGS. 4 a - d are block diagrams that illustrate example execution graphs.

FIG. 2 illustrates an example of a continuous processing system that processes streaming data. The system 200 includes a programming interface 210 , a compiler 220 , and a continuous processing engine 230 , where the engine may be distributed over a cluster of servers. The programming interface 210 enables programmers to write statements 260 (such as queries), which are then complied by the compiler 220 and provided to the continuous processing engine 230 .

A coherent sequence of statements is referred to as a “module.” A module subscribes to one or more input data streams 240 and publishes to one or more output data streams 250 . Through the engine 230 , a module operates on input data streams continuously.

When the compiler 220 compiles a module, an execution graph is created for executing the module. The execution graph comprises a set of connected primitives, where the primitives correspond to statements in the module. (Note: Such correspondence is usually not a one-to-one correspondence.) Examples of primitives include filters, joiners, aggregators, and windows.

Coral8, Inc.'s “Complex Event Processing” engine is an example of a continuous processing system. Also, one embodiment of a continuous processing system is described in U.S. patent application Ser. No. 11/015,963, filed on Dec. 17, 2004 with Mark Tsimelzon as the first-named inventor, and titled “Publish and Subscribe Capable Continuous Query Processor for Real-time data streams,” the contents of which are incorporated by reference as if fully disclosed herein.

Statements may be written in a continuous-processing software language (CPL), which is sometimes also referred to as a continuous correlation language (CCL). An example of such a language described in the U.S. patent application Ser. No. 11/346,119, filed on Feb. 2, 2006, and titled “Continuous Processing Language for Real-time Data Streams,” the contents of which are incorporated by reference as if fully disclosed herein.

In cases where an execution graph in a continuous processing system includes primitives with potential for substantial delay, the present invention provides a system and method for synchronizing the processing of messages in areas of the execution graph affected by (i.e., dependent on) primitives with potential for substantial delay. FIG. 3 illustrates one embodiment of such method. In the preferred embodiment, the steps illustrated in FIG. 3 are performed by a compiler (e.g., compiler 220 ) in the continuous processing system, but the invention is in no way limited to the steps being performed by the compiler. The steps of FIG. 3 are described with reference to the example execution graphs in FIG. 4 a - d.

In creating the execution graph, the compiler identifies primitives with potential for substantial delay (step 310 ). In one embodiment, the compiler has a hard-coded list of primitives with potential for substantial delay. Alternatively, the compiler may perform such identification based on rules or settings entered by a programmer or administrator for the continuous processing system. As stated above, examples of primitives with potential for substantial delay include database joiner primitives, primitives that make remote procedure calls, and primitives that are user-defined functions that can take substantial time to complete.

For each primitive with potential for substantial delay, the compiler determines whether such primitive is within a loop (step 320 ). For each such primitive that is not within a loop, the compiler designates all primitives downstream of such primitive as a “timezone” (step 330 ). The output(s) of the potentially-delayed primitive is an input to the timezone. A “timezone” is an area of the execution graph that, at any given time, only includes messages that need to be processed together. In one embodiment, this means that, at a given time, a timezone includes only messages with the same timestamp. Messages enter a timezone synchronously.

In FIG. 4 a , which illustrates an example execution graph, database joiner 410 is a primitive with potential for substantial delay. Therefore, in creating this execution graph 400 , the compiler designates all primitives 425 downstream of database joiner 410 as being part of a timezone 420 . The outputs of database joiner 410 are inputs to the timezone 420 . “P” in FIGS. 4 a - c stands for primitive.

The compiler then identifies all other inputs, if any, to the timezone that are in addition to the input(s) from the potentially-delayed primitive (step 340 ). The compiler places a synchronizer across all inputs to the timezone, where the synchronizer controls when messages on such inputs enter the timezone to ensure that messages that need to be processed together enter the timezone at the same time (step 350 ). The synchronizer does not let additional messages into the timezone until the previously-entered messages have been processed by the primitives in the timezone.

In the execution graph in FIG. 4 b , the output of primitive 490 is an input to the timezone 420 . The compiler places a synchronizer 450 across all inputs to the timezone (namely, the outputs of database joiner 410 and the output of primitive 490 ). The use of timezone 420 enables database joiner 410 and primitives 480 , 490 , and 495 to still process messages in parallel (such as messages with different internal timestamps), while protecting primitives 425 from such parallel processing, as parallel processing could cause such primitives 425 to process messages out of order due to delays associated with database joiner 410 . Timezone 420 and synchronizer 450 enable database joiner 410 to make concurrent calls to the database (instead of serial calls) without disturbing the order in which messages downstream of database joiner 410 are processed.

If a primitive with potential for substantial delay is within a loop, the compiler designates such primitive and all primitives downstream of such primitive as a timezone (step 350 ). This differs from the non-loop scenario (steps 330 - 340 ) in that the timezone includes the potentially-delayed primitive. The compiler then identifies all inputs to the timezone (step 360 ) and places a synchronizer across all inputs to the timezone, where the synchronizer controls when messages on such inputs enter the timezone to ensure that messages that need to be processed together enter the timezone at the same time (step 370 ). The synchronizer does not let additional messages into the timezone until the previously-entered messages have been processed by the primitives in the timezone.

In the example execution graph in FIG. 4 c , database joiner 410 is a primitive with potential for substantial delay. Since database joiner 410 is part of a loop, timezone 420 includes database joiner 410 , as well as all primitives 425 downstream of database joiner 410 . Synchronizer 450 is placed across all inputs to the timezone 420 .

In one embodiment, the synchronizer relies on the internal timestamp (not the server clock time) of messages to determine which messages go into the timezone at a particular time. In such embodiment, this means that messages with the same timestamp go into the timezone at the same time, and messages with the next timestamp are not allowed into the timezone until messages with the previous timestamp have been processed in the timezone.

In an alternate embodiment, incoming messages are divided into “time slices.” A time slice is a set of messages that have the same time slice and that are processed together. In this embodiment, the synchronizer ensures that messages with the same time slice enter the timezone at the same time. Only one time slice is allowed in the timezone at a time. Time slices are described in more detail in U.S. patent application Ser. No. 11/787,912, filed on Apr. 18, 2007 with inventors Aleksey Sanin, Mark Tsimelzon, Ian D. Marshall, and Robert B. Hagmann and titled “Method for Obtaining Repeatable and Predicable Output Results in a Continuous Processing System,” the contents of which are incorporated by reference as if fully described herein.

Steps 320 - 340 or steps 350 - 370 , whichever is applicable, are performed for each primitive in the graph with potential for substantial delay. Timezones can be nested within other timezones. For example, in FIG. 4 d , the execution graph 400 includes a database joiner 460 that is downstream of database joiner 410 . Timezone 470 is associated with database joiner 460 and is nested within timezone 420 . A synchronizer 475 is placed across all inputs to timezone 470 .

The use of timezones allows certain areas of the execution graph (i.e., areas outside the timezone) to process messages with different timestamps concurrently, while protecting other areas of the graph from processing messages out of order due to primitives with potential for substantial delay. This is especially useful in cases where it desirable for a database joiner primitive (that is not in a loop) to make concurrent calls to a database for multiple joins while still maintaining the proper order in which messages downstream of the database joiner 410 should be processed.

As will be understood by those familiar with the art, the invention may be embodied in other specific forms without departing from the spirit or essential characteristics thereof. Accordingly, the above disclosure of the present invention is intended to be illustrative and not limiting of the invention.