Smart tuple condition-based operation performance

A stream application receives a stream of tuples to be processed by a plurality of processing elements operating on one or more compute nodes. Each processing element has one or more stream operators. The stream application assigns one or more processing cycles to one or more segments of software code. The segments of software code are embedded in one or more tuples of the stream of tuples. The stream application determines that a condition is met based on the processing cycles. The stream application performs an operation based on the processing cycles.

BACKGROUND

The present disclosure relates to stream computing, and more specifically, to performing an operation based upon a condition tested by a smart tuple in a smart stream computing environment.

Stream computing may be utilized to provide real-time analytic processing to large quantities of data. Stream computing may be used for scientific research purposes, such as weather forecasting and complex physics modelling. Stream computing may be used for commercial purposes, such as real-time inventory management and stock market tracking. Stream computing may be used for medical purposes, such as analyzing complex and interconnected functions of the human body. Stream computing may be used by end users to more immediately and accurately understand and contextualize large amounts of information.

SUMMARY

According to embodiments of the present disclosure, disclosed herein is a method for processing a stream of tuples. A stream application receives a stream of tuples to be processed by a plurality of processing elements operating on one or more compute nodes. Each processing element has one or more stream operators. The stream application assigns one or more processing cycles to one or more segments of software code. The segments of software code are embedded in one or more tuples of the stream of tuples. The stream application determines that a condition is met based on the processing cycles. The stream application performs an operations based on the processing cycles.

Also disclosed herein are embodiments of a system for processing a stream of tuples. The system includes a plurality of processing elements configured to receive a stream of tuples. Each processing element has one or more stream operators. The system also includes two or more processors. The system further includes a memory containing an application that causes at least one of the two or more processors to perform a method. A first processor embeds one or more tuples of the stream of tuples with one or more segments of software code. A second processor determines, based on the one or more segments of software code, that a condition is met. The second processor performs, based on the determination and the one or more segments of software code, an operation.

Also disclosed herein are embodiments of a computer program product for processing a stream of tuple. The computer program product includes a computer readable storage medium having program instructions executable by a plurality of processing elements operating on one or more compute nodes. Each processing element has one or more stream operators. A first compute node embeds one or more tuples of the stream of tuples with one or more segments of software code. A second compute node determines, based on the one or more segments of software code, that a condition is met. The second compute node performs, based on the determined condition and based on the one or more segments of software code, an operation.

DETAILED DESCRIPTION

Aspects of the present disclosure relate to stream computing, more particular aspects relate to performing an operation based upon a condition tested by a smart tuple in a smart streams computing environment. While the present disclosure is not necessarily limited to such applications, various aspects of the disclosure may be appreciated through a discussion of various examples using this context.

One of the primary uses of computing systems (alternatively, computer systems) is to collect available information, manipulate the collected information, and make decisions based on the manipulated information. Existing computer systems may operate on information through means of databases that allow users to determine what has happened and to make predictions for future results based on past events. These computer systems receive information from a variety of sources and then record the information into permanent databases. After the information has been recorded in the databases, the computing systems run algorithms on the information sometimes generating new information and then performing associated transformations on and storing of the new information—to make determinations and provide context to users.

The ability of these existing computer systems to analyze information and provide meaning to users may be insufficient in some situations. The ability of large organizations, such as corporations and governments, to make decisions based on information analysis may be impaired by the limited scope of the information available. In addition, the analysis may be of limited value because it relies on stored structural databases that may contain out-of-date information. This may lead to decisions that are of limited value or, in some cases, inaccurate. For example, a weather forecast service may be unable to accurately predict precipitation for a given region, or a stock brokerage firm may make an incorrect decision regarding a trend in trading of shares.

The analytical shortcomings of existing computer systems may be compounded by other factors. First, the world is becoming more instrumented, as previously unintelligent devices are now becoming intelligent devices. Intelligent devices may include devices that have historically been unable to provide analytical information but with the additions of sensors can now do so (e.g., automobiles that are now able to provide diagnostic information to their owners or manufacturers, thermostats that now communicate information about daily temperature fluctuations in homes to users via webpages). Second, these shortcomings may also be compounded by an increase in communication from information sources, as previously isolated devices are now becoming interconnected (e.g., appliances within homes communicate with each other and with power utilities to more efficiently utilize electricity). These new sources of information may provide volumes of not only isolated data points but also relationships between the newly intelligent devices.

A third compounding factor is that users of computing systems may desire continuous analysis of streams of information, while current methods of data acquisition may provide only an event-based approach of analyzing pre-recorded information. For example, an existing analytics package may receive a finite amount of data and, later, apply analysis to the data. This approach may not work when dealing with a continuous stream of data. A fourth compounding factor is that existing computer systems may have deficiencies in handling not only the volume of information but also in dealing with the unstructured nature of the information; for example, sensors, cameras, and other new data sources may provide no context or format, just raw information. The existing analytics methods of conventional computing systems may need to modify and rearrange this data in order to provide any kind of context for the raw information. The modifications and rearrangements may take time or resources that many existing computing systems may not be able to provide.

Yet another potential drawback is that existing computing systems may not provide scalable solutions to new users. The advent of smart and connected devices has provided new use-cases for analytics of continuous streams of information. Modern systems of large-scale data collection, however, may require significant user training and provide unintuitive interfaces. For example, a farmer may have each animal on a farm instrumented with sensors to monitor the health and location of the animals. The data from these sensors may enable the farmer to respond to ever-changing health conditions of the animals, but only if the sensor data is collected and transformed into a usable format to provide meaningful information to the farmer in real-time. The farmer may not have the money to provide training and resources to a technical expert to construct a large-scale analytics package, and the obtained information may be left used.

I. Stream Computing

Stream-based computing (e.g., within a stream application) may provide users with a way to obtain meaning from extremely large sets of information (big-data). Stream computing may provide users with the ability to analyze information as it is captured but before it reaches a final destination (e.g., data from sensors being transmitted to a flat file, records being collected from internet queries and being stored to a database). In some embodiments, stream computing may provide users with the ability to analyze a stream of information that is too large to be captured and placed into a final destination (e.g., sensor values from thousands of sensors that will be discarded after being measured could be utilized by a stream computing application to provide detailed analysis). Stream computing may provide the bandwidth to process big-data continuously and in real-time (e.g., generating context from tens of millions of records per second with low latency from record reception to provide meaningful action in microseconds). Stream computing may provide users with the ability to utilize familiar programmatic conventions to provide context to big-data (e.g., using a structured language to retrieve, format, and conditionally select a subset of information regarding millions of records as those records are generated, using conditional language to trigger an action every few milliseconds based on traditional program statements applied every hundred microseconds).

Information flowing through a stream application may be in the form of streams. A stream may be made up of one or more tuples. A tuple may be a sequence of one or more associated attributes in a relational format. The tuples may share characteristics of a classical relational database (e.g., a single tuple may be similar to a row in a relational database and the attributes of a tuple may be similar to the columns of the row). The tuples may have non-relational database relationships to other tuples of a stream application (e.g., individual values, key-value pairs, flat files, etc.). Tuples may include values in a variety of known computer formats (e.g., integer, float, Boolean, string, etc.). Tuples may contain attributes about themselves, such as metadata. As used herein, a stream, streams, or data stream may refer to a sequence of tuples flowing through a stream application. Generally, a stream may be considered a pseudo-infinite sequence of tuples.

FIG. 1depicts a stream computing application (stream application)100consistent with embodiments of the present disclosure. The stream application100may be represented in the form of an operator graph102. The operator graph102may visually represent to a user the flow of data through the stream application100. The operator graph102may define how tuples are routed through the various components of the stream application100(e.g., an execution path). The stream application100may include one or more compute nodes110A,110B,110C, and110D (collectively,110); a development system120; a management system130; one or more processing elements140A,140B,140C,140D,140E, and140F (collectively,140); and one or more stream operators142A,142B,142C,142D,142E,142F,142G (collectively,142). The stream application100may receive information from one or more sources144and may output information to one or more sinks146.

It should be appreciated that the stream application100depicted inFIG. 1is for example purposes. Stream applications may vary in the number of compute nodes, processing elements, or stream operators. The stream application may also vary the roles and/or responsibilities performed by any of the components or may include other components not depicted. For example, some or all of the functionality of the development system120may be performed by the management system130. In another example, the functionalities of the development system120and the management system130may be performed by a singular administrative system (not depicted). The administrative system may be configured to perform these tasks without deviation from the embodiments disclosed herein. In yet another example, the functionalities of the development system120and the management system130may be performed by a plurality of services (e.g., ten or more individual software programs each configured to perform a specific function).

The compute nodes110may be computer systems and may each include the following components: a processor, a memory, and an input/output interface (herein I/O). Each compute node110may also include an operating system or a hypervisor. In some embodiments, the compute nodes110may perform operations for the development system120, the management system130, the processing elements140, and/or the stream operators142. The compute nodes110may be categorized as management hosts, application hosts, or mixed-use hosts. A management host may perform operations for the development system120and/or the management system130. An application host may perform operations for the processing elements140and stream operators142. A mixed-use host may perform operations of both a management host and an application host.FIG. 5depicts a computer system that may be a compute node consistent with embodiments of the present disclosure.

A network (not depicted) may commutatively couple each of the nodes110together (e.g., a local area network, the Internet, etc.). For example, node110A may communicate with nodes110B,110C, and110D through the network. The computes nodes110may communicate with the network by way of the I/O. The network may include a variety of physical communication channels or links. The links may be wired, wireless, optical, or any other suitable media. The network may include a variety of network hardware and software for performing routing, switching, and other functions, such as routers, switches, or bridges. The nodes110may communicate through a variety of protocols (e.g., the internet protocol, the transmission control protocol, the file transfer protocol, the hypertext transfer protocol, etc.). In some embodiments, the nodes110may share the network with other hardware, software, or services (not depicted).

The development system120may provide a user with the ability to create a stream application that is targeted to process specific sets of data. The development system120may operate on a computer system (not depicted), such as the computer system depicted inFIG. 5. The development system120may operate on one or more of the compute nodes110. The development system120may generate one or more configuration files that describes the stream computing application100(e.g., the processing elements140, the stream operators142, the sources144, the sinks146, the assignment of the aforementioned to the compute nodes110, etc.). The development system120may receive requests from a user to generate the stream application100. The development system120may receive requests from a user to generate other stream applications (not depicted). The development system120may communicate with the management system130to pass along the configuration on any stream applications it creates.

The development system120may generate the configuration by considering the performance characteristics of the software components (e.g., the processing elements140, the stream operators142, etc.) the hardware (e.g., the compute nodes110, the network) and the data (e.g. the sources144, the format of the tuples, etc.). In a first example, the development system120may determine that the overhead of running processing elements140A,140B, and140C together on compute node110A results in better performance than running them on separate compute nodes. The performance may be better because of a latency incurred by running processing elements140A,140B, and140C across the network between compute nodes110A and110B. In a second example, the development system120may determine that the memory footprint of placing stream operators142C,142D,142E, and142F into a single processing element140E is larger than the cache of a first processor in compute node110B. To preserve memory space inside the cache of the first processor the development system120may decide to place only the stream operators142D,142E, and142F into a single processing element140E despite the inter-process communication latency of having two processing elements140D and140E.

In a third example of considering the performance characteristics, the development system120may identify a first operation (e.g., an operation being performed on processing element140F on compute node110C) that requires a larger amount of resources within the stream application100. The development system120may assign a larger amount of resources (e.g., operating the processing element140F on compute node110D in addition to compute node110C) to aid the performance of the first operation. The development system120may identify a second operation (e.g., an operation being performed on processing element140A) that requires a smaller amount of resources within the stream application100. The development system120may further determine that the stream application100may operate more efficiently through an increase in parallelization (e.g., more instances of processing element140A). The development system120may create multiple instances of processing element140A (e.g., processing elements140B and140C). The development system120may then assign processing elements140A,140B, and140C to a single resource (e.g., compute node110A). Lastly, the development system120may identify a third operation and fourth operation (e.g., operations being performed on processing elements140D and140E) that each require low levels of resources. The development system120may assign a smaller amount of resources to the two different operations (e.g., having them share the resources of compute node110B rather than each operation being performed on its own compute node).

The development system120may include a compiler (not depicted) that compiles modules (e.g., processing elements140, stream operators142, etc.). The modules may be source code or other programmatic statements. The modules may be in the form of requests from a stream processing language (e.g., a computing language containing declarative statements allowing a user to state a specific subset from information formatted in a specific manner). The compiler may translate the modules into an object code (e.g., a machine code targeted to the specific instruction set architecture of the compute nodes110). The compiler may translate the modules into an intermediary form (e.g., a virtual machine code). The compiler may be a just-in-time compiler that executes as part of an interpreter. In some embodiments, the compiler may be an optimizing compiler. In some embodiments, the compiler may perform peephole optimizations, local optimizations, loop optimizations, inter-procedural or whole-program optimizations, machine code optimizations, or any other optimizations that reduce the amount of time required to execute the object code, to reduce the amount of memory required to execute the object code, or both.

The management system130may monitor and administer the stream application100. The management system130may operate on a computer system (not depicted), such as the computer system depicted inFIG. 5. The management system130may operate on one or more of the compute nodes110. The management system130may also provide the operator graph102of the stream application100. The management system130may host the services that make up the stream application100(e.g., services that monitor the health of the compute nodes110, the performance of the processing elements140and stream operators142, etc.). The management system130may receive requests from users (e.g., requests to authenticate and authorize users of the stream application110, requests to view the information generated by the stream application, requests to view the operator graph102, etc.).

The management system130may provide a user with the ability to create multiple instances of the stream application100configured by the development system120. For example, if a second instance of the stream application100is required to perform the same processing, then the management system130may allocate a second set of compute nodes (not depicted) for performance of the second instance of the stream application. The management system130may also reassign the compute nodes110to relieve bottlenecks in the system. For example, as shown, processing elements140D and140E are executed by compute node110B. Processing element140F is executed by compute nodes110C and110D. In one situation, the stream application100may experience performance issues because processing elements140D and140E are not providing tuples to processing element140F before processing element140F enters an idle state. The management system130may detect these performance issues and may reassign resources from compute node110D to execute a portion or all of processing element140D, thereby reducing the workload on compute node110B. The management system130may also perform operations of the operating systems of the compute nodes110, such as the load balancing and resource allocation of the processing elements140and stream operators142. By performing operations of the operating systems, the management system130may enable the stream application100to more efficiently use the available hardware resources and increase performance (e.g., by lowering the overhead of the operating systems and multiprocessing hardware of the compute nodes110).

The processing elements140may perform the operations of the stream application100. Each of the processing elements140may operate on one or more of the compute nodes110. In some embodiments, a given processing element140may operate on a subset of a given compute node110, such as a processor or a single core of processor of a compute node110. In some embodiments, a given processing element140may operate on multiple compute nodes110. The processing elements140may be generated by the development system120. Each of the processing elements140may be in the form of a binary file and additionally library files (e.g., an executable file and associated libraries, a package file containing executable code and associate resources, etc.).

Each of processing elements140may include configuration information from the development system120or the management system130(e.g., the resources and conventions required by the relevant compute node110to which it has been assigned, the identity and credentials necessary to communicate with the sources144or sinks146, the identity and credentials necessary to communicate with other processing elements, etc.). Each of the processing elements140may be configured by the development system120to run optimally upon one of the compute nodes110. For example, processing elements140A,140B, and140C may be compiled to run with optimizations recognized by an operating system running on compute node110A. The processing elements140A,140B, and140C may also be optimized for the particular hardware of compute node110A (e.g., instruction set architecture, configured resources such as memory and processor, etc.).

Each of processing elements140may include one or more stream operators142that perform basic functions of the stream application100. As streams of tuples flow through the processing elements140, as directed by the operator graph102, they pass from one stream operator to another. Multiple stream operators142within the same processing element140may benefit from architectural efficiencies (e.g., reduced cache missed, shared variables and logic, reduced memory swapping, etc.). The processing elements140and the stream operators142may utilize inter-process communication (e.g., network sockets, shared memory, message queues, message passing, semaphores, etc.). The processing elements140and the stream operators142may utilize different inter-process communication techniques depending on the configuration of the stream application100. For example: stream operator142A may use a semaphore to communicate with stream operator142B; processing element140A may use a message que to communicate with processing element140C; and processing element140B may use a network socket to communicate with processing element140D.

The stream operators142may perform the basic logic and operations of the stream application100(e.g., processing tuples and passing processed tuples to other components of the stream application). By separating the logic that would conventionally occur within a single larger program into basic operations performed by the stream operators142, the stream application100may provide greater scalability. For example, tens of compute nodes hosting hundreds of stream operators in a stream application may enable processing of millions of tuples per second. The logic may be created by the development system120before runtime of the stream application100. In some embodiments, the sources144and the sinks146may also be stream operators142. In some embodiments, the sources144and the sinks146may link multiple stream applications together (e.g., the sources144could be sinks for a second stream application and the sinks146could be sources for a third stream application). The stream operators142may be configured by the development system120to optimally perform the stream application100using the available compute nodes110. The stream operators may142send and receive tuples from other stream operators. The stream operators142may receive tuples from the sources144and may send tuples to the sink146.

The stream operators142may perform operations (e.g., conditional logic, iterative looping structures, type conversions, string formatting, etc.) upon the attributes of a tuple. In some embodiments, each stream operator142may perform only a very simple operation and may pass the updated tuple on to another stream operator in the stream application100—simple stream operators may be more scalable and easier to parallelize. For example, stream operator142B may receive a date value to a specific precision and may round the date value to a lower precision and pass the altered date value to stream operator142D that may change the altered date value from a 24-hour format to a 12-hour format. A given stream operator142may not change anything about a tuple. The stream operators142may perform operations upon a tuple by adding new attributes or removing existing attributes.

The stream operators142may perform operations upon a stream of tuples by routing some tuples to a first stream operator and other tuples to a second stream operator (e.g., stream operator142B sends some tuples to stream operator142C and other tuples to stream operator142D). The stream operators142may perform operations upon a stream of tuples by filtering some tuples (e.g., culling some tuples and passing on a subset of the stream to another stream operator). The stream operators142may also perform operations upon a stream of tuples by routing some of the stream to itself (e.g., stream operator142D may perform a simple arithmetic operation and as part of its operation it may perform a logical loop and direct a subset of tuples to itself). In some embodiments, a particular tuple output by a stream operator142or processing element140may not be considered to be the same tuple as a corresponding input tuple even if the input tuple is not changed by the stream operator or the processing element.

II. Smart Stream Computing

Stream computing may allow users to process big-data and provide advanced metrics upon that big-data continuously as it is being generated by a variety of sources. A stream application may provide stream computing by generating a configuration of one or more processing elements, each processing element containing one or more stream operators. Each processing element and/or stream operator of the stream application may process big-data by generating and modifying information in the form of tuples. Each tuple may have one or more attributes (e.g., the tuples may be analogous to rows and the attributes analogous to columns in a table).

The stream application may deploy an instance of the configuration to a set of hardware compute nodes.FIG. 5depicts a computer system that may be a compute node consistent with embodiments of the present disclosure. The stream application may then administer the instance by adjusting the hardware to perform the stream application as it is configured, such as by load balancing the processing elements onto compute nodes, onto a portion of a given compute node, or across multiple compute nodes.

In some situations, a stream application may be largely a static big-data operating mechanism. Such a stream application once configured may not be changeable in the context it provides to a user. Further, in some situation, such a stream application performs certain logic in how it processes tuples. This logic once configured may not be updatable or changeable until a new stream application is compiled. Trying to provide an update to a processing element or stream operator of such a configured stream instance may be impractical because of the real-time continuous nature of stream applications and the information stream applications process. For example, any down-time, even in microseconds, may cause the stream application to not collect one or more tuples during the changeover from an originally configured processing element to an updated processing element. Missing a portion of the data may provide a partial or complete failure of the stream application and may result in the stream application being unable to provide users with context to big-data sources.

Choosing not to update the configuration of a stream application may also be undesirable because the configured logic may have faults or assumptions. For example, a user may be using an instance of a stream application to monitor weather from hundreds of weather sensors across many locations to better and more accurately guide and aim solar panels. If the user provided an error in the logic of the stream application or utilized an out-of-date set of metrics when the stream application was configured, the stream application may provide meaningless context. Such a misconfigured stream application may discard portions of meaningful tuples from the weather sensors, and without a way to alter the logic of the stream application while it is running, these tuples may be lost.

Associating a segment of code with one or more tuples may create a stream application with enhanced flexibility (smart stream application). A stream application may operate upon one or more tuples that contain attributes (e.g., tuples flow through pathways and are altered in some way by one or more stream operators and are sent along more pathways from those stream operators to other stream operators). A smart stream application may also have one or more code-embedded tuples (smart tuples)—a code-embedded tuple or smart tuple may also be referred to as an embedded tuple. The smart tuples may add further programming logic to a stream application by adding additional intelligence outside of the stream operators (e.g., adding processing power to the pathways by way of the tuples). The smart stream application may be able to dynamically modify the level of tuple processing power as resources allow (e.g., only a few tuples may be smart tuples during high usage, a large amount of tuples may be smart tuples during low usage, all or none of the tuples may be smart tuples, etc.). The smart stream application may alter the tuple processing power without upsetting the performance of the stream application (e.g., additional hardware may be added for processing smart tuples).

The smart tuples may have additional capabilities not found in normal tuples (e.g., know its own position in the stream application, communicate to other tuples, communicate with the administrative components of the stream application, communicate with components external to the stream application, etc.). The smart tuples may also provide additional flexibility to the stream application (e.g., changing the logic of the stream application by a smart tuple bypassing one or more processing elements and/or stream operators, adding increased logic during low volumes of data by providing additional operations through the smart tuple in between processing elements and/or stream operators). A smart stream application may also be updated by one or more smart tuples (e.g., a smart tuple may contain an update or patch).

In a first example, functionality for processing tuples within a first stream operator may be set to a specific formula. By utilizing smart tuples, a user could update the functionality through a smart tuple having an altered formula and an update script to enact the altered formula. The stream operator may receive the alteration to the formula from the update script and may begin processing tuples based on the altered formula. In a second example, a temporary change of functionality could occur through the use of multiple smart tuples. A second stream operator may perform a set action on a stream of tuples. Each of the multiple smart tuples may be encoded to perform an updated action on one tuple from the stream of tuples. The multiple smart tuples may also reroute the stream of tuples, thus bypassing the second stream operator. As long as the smart stream application provides processing of tuples to the smart tuples instead of the second stream operator the updated action may occur upon the stream of tuples. In a third example, a temporary addition of functionality could occur through the use of multiple smart tuples. A third stream operator may perform calculations and update attributes from a first subset of a stream of tuples. Each of the multiple smart tuples may be encoded to perform the calculations on a subset of the stream of tuples not updated by the third stream operator. As long as the smart stream application provides processing of tuples to the smart tuples in addition to the third stream operator an increased level of detail may occur upon the stream of tuples—more tuples from the stream of tuples may have updated attributes.

FIG. 2depicts a smart stream application200consistent with embodiments of the present disclosure. The smart stream application200may be represented in the form of an operator graph202. The operator graph202may visually represent to a user the flow of data through the smart stream application200. The operator graph202may define how tuples are routed through the various components of the smart stream application200(e.g., an execution path). The smart stream application200may include one or more compute nodes210A,210B,210C, and210D (collectively,210); a development system220; a management system230; one or more processing elements240A,240B,240C,240D,240E, and240F (collectively,240); and one or more stream operators242A,242B,242C,242D,242E,242F,242G (collectively,242). The smart stream application200may receive information from a source244and may output information to a sink246. The source244and the sink246may be stream operators. The compute nodes210may be communicatively coupled to each other through a network (not depicted). The smart stream application200may also include one or more processing element tuple executors (PETEs)250A,250B,250C,250D,250E, and250F (collectively250); and one or more stream operator tuple executors (SOTEs)255. The functionality of a PETE could be replicated by one or more SOTEs (e.g., PETE250E could be replicated by a SOTE within stream operators242D,242E, and242F).

The compute nodes210may be one or more physical or virtual computers that are configured to enable execution of the other components of the smart stream application200.FIG. 5depicts a computer system that may be a compute node consistent with embodiments of the present disclosure. The development system220may enable the smart stream application to generate the operator graph202based on a request from the user. The development system220may receive from the user a request to perform some kind of structure-language query (e.g., select a subset of readings from hundreds of vitality sensors in a dozen hospitals based on a complex criteria continuously throughout a month, and, as the millions of readings in the subset are selected, format them in a certain arrangement, perform subtotaling and generate periodic notifications, etc.). The development system220may assess the available compute nodes210and generate the operator graph202(e.g., the layout and arrangement of the processing elements240and stream operators242). The management system230may monitor the smart stream application200as it operates and provide management capabilities such as reassigning compute nodes210to alleviate bottlenecks.

The smart stream application200may be configured to process tuples (each tuple being an association of one or more attributes) collected from the source244and deposit the processed tuples in the sink246. In detail, the source244may generate tuples that flow to the processing elements240A,240B,240C. The processing elements240A,240B, and240C may receive the tuples and generate a second and third set of tuples—then processing elements240A,240B, and240C may send the second and third sets of tuples to processing elements240D and240E, respectively. The processing element240D and may generate a fourth set of tuples from the second set of tuples and pass the fourth set of tuples onto processing element240F. The processing element240E may generate a fifth set of tuples from the third set of tuples and pass the fifth set of tuples onto processing element240F. Finally processing element240F may generate a sixth set of tuples and pass the sixth set of tuples onto the sink246. In each of the processing elements240the stream operators242may perform the alterations to the tuples (e.g., adding or removing attributes, generating new attributes, determining the route of tuples, adding new tuples, removing existing tuples, etc.). In some embodiments, the stream operators242may pass tuples to each other within a given processing element240(e.g., stream operators242A and242B within processing element240A).

The PETEs250and SOTEs255may be configured to enable the creation and processing of the smart tuples270A,270B,270C,270D,270E,270F,270G,270H,270I (collectively,270). The management system230may also be configured to enable the creation and processing of the smart tuples270in the smart stream application200. In detail, the management system230may enable smart stream operation by sending a command to the source244along with one or more segments of code. The SOTE255may generate the smart tuples270by wrapping them with the segments of code (e.g., adding attributes to the tuples that contain a code object, adding attributes to the tuples that contain a link to a code object, etc.). The code objects may also be added to the compute nodes210such that they are accessible by processing elements240and stream operators242. The management system230may also enable smart stream operation by sending a command to the processing elements240.

The processing elements240in response to the management system230may instruct the PETEs250to detect smart tuples270and may provide access to processing cycles of the compute nodes210to the segments of code wrapped in the smart tuples270. The PETEs250and SOTEs255may receive access to processing cycles periodically (e.g., every nanosecond, every three operations of a given stream operator242, every ten operations of a given processing element240, etc.). The PETEs and SOTEs may receive access to the processing cycles in another manner (e.g., before execution of a given stream operator242, after execution of a given stream operator, etc.). The processing elements240and stream operators242may preserve the smart tuples270as they receive tuples, process the received tuples, and generate new tuples. For example, during the processing of tuples stream operator242C may generate new tuples (e.g., perform some processing and create a new tuple based on the result). Smart tuple270C may be processed by stream operator242C upon entering processing element240D. During generation of a new tuple based on smart tuple270C, the stream operator may wrap the new tuple with the same segment of code that was wrapped with smart tuple270C.

The management system230may be configured to disable the smart stream operation of the smart stream application200. The management system230may disable smart stream operation by searching for each of the smart tuples270and unwrapping the segments of code (e.g., removing attributes from the tuples that contain a code object, removing attributes from the tuples that contain a link to a code object, etc.). In some embodiments, the management system230may disable smart stream operation by sending signals to the processing elements240, the stream operators242, and/or the source244to ignore the wrapped segments of code.

FIG. 3depicts a stream application300with smart stream capabilities consistent with embodiments of the present disclosure. The stream application300may be represented in the form of an operator graph302. The operator graph302may define how tuples are routed through the various components of the stream application300(e.g., an execution path). The stream application300may include one or more compute nodes310A,310B,310C, and310D (collectively,310); a development system320; a management system330; one or more processing elements340A,340B,340C,340D,340E, and340F (collectively,340); and one or more stream operators342A,342B,342C,342D,342E,342F,342G (collectively,342). The stream application300may receive information from a source344and may output information to a sink346. The source344and the sink346may be stream operators. The compute nodes310may be communicatively coupled to each other through a network (not depicted). The stream application300may also include a tuple intelligence manager360(TIM). The TIM360may commutatively couple to the network and may be configured to interact with the other components of the stream application300.

The compute nodes310may be one or more physical or virtual computers that are configured to enable execution of the other components of the stream application300.FIG. 5depicts a computer system that may be a compute node consistent with embodiments of the present disclosure. The development system320may enable the stream application to generate the operator graph302based on a request from the user. The development system320may receive from the user a request to perform some kind of structure-language query (e.g., select a subset of readings from hundreds of vitality sensors in a dozen hospitals based on a complex criteria continuously throughout a month, and, as the millions of readings in the subset are selected, format them in a certain arrangement, perform subtotaling and generate periodic notifications, etc.). The development system320may assess the available compute nodes310and generate the operator graph302(e.g., the layout and arrangement of the processing elements340and stream operators342). The management system330may monitor the stream application300as it operates and provide management capabilities such as reassigning compute nodes310to alleviate bottlenecks.

The stream application300may be configured to process tuples (each tuple being an association of one or more attributes) collected from the source344and deposit the processed tuples in the sink346. In detail, the source344may generate tuples that flow to the processing elements340A,340B,340C. The processing elements340A,340B, and340C may receive the tuples and generate a second and third set of tuples—then processing elements340A,340B, and340C may send the second and third sets of tuples to processing elements340D and340E, respectively. The processing element340D and may generate a fourth set of tuples from the second set of tuples and pass the fourth set of tuples onto processing element340F. The processing element340E may generate a fifth set of tuples from the third set of tuples and pass the fifth set of tuples onto processing element340F. Finally processing element340F may generate a sixth set of tuples and pass the sixth set of tuples onto the sink346. In each of the processing elements340the stream operators342may perform the alterations to the tuples (e.g., adding or removing attributes, generating new attributes, determining the route of tuples, adding new tuples, removing existing tuples, etc.). In some embodiments, the stream operators342may pass tuples to each other within a given processing element340(e.g., stream operators342A and342B within processing element340A).

The TIM360may be configured to enable the creation and processing of the smart tuples370A,370B,370C,370D,370E,370F,370G,370H,370I (collectively,370). In detail, the TIM360may enable smart stream operation by generating the smart tuples370. The smart tuples370may be generated by wrapping them with the segments of code (e.g., adding attributes to the tuples that contain a code object, adding attributes to the tuples that contain a link to a code object, etc.). The code objects may also be added to the compute nodes310such that they are accessible by processing elements340and stream operators342.

The TIM360may provide access to processing cycles of the compute nodes310to the segments of code wrapped in the smart tuples370. In some embodiments, the TIM360may enable smart stream operation by providing access to processing cycles of additional computing systems (not depicted). The TIM360may provide access to processing cycles periodically (e.g., every nanosecond, every three operations of a given stream operator342, every ten operations of a given processing element340, etc.). The TIM360may provide access to the processing cycles in another manner (e.g., before execution of a given stream operator342, after execution of a given stream operator, etc.).

The TIM360may preserve the order of tuples in the stream application300. In detail, while the TIM360is providing a given smart tuple370access to processing cycles, a given processing element340and/or stream operator342may be preparing to process the given smart tuple. The TIM360may prevent the given processing element340and/or stream operator342from processing the given smart tuple370by issuing a wait command to the given processing element and/or stream operator (either directly or through a request to the management system330). In response to the wait command the given processing element340and/or stream operator342may pause operation until the given smart tuple370finishes.

The processing elements340and stream operators342may preserve the smart tuples370as they receive tuples, process the received tuples, and generate new tuples. For example, during the processing of tuples stream operator342C may generate new tuples (e.g., perform some processing and create a new tuple based on the result). Smart tuple370C may be processed by stream operator342C upon entering processing element340D. During generation of a new tuple based on smart tuple370C, the stream operator may wrap the new tuple with the same segment of code that was wrapped with smart tuple370C. In some embodiments, the TIM360may monitor the stream operators342and the processing elements340and may preserve the smart tuples370as new tuples are generated.

The TIM360may be configured to disable the smart stream operation of the stream application300. The TIM360may disable smart stream operation by searching for each of the smart tuples370and unwrapping the segments of code (e.g., removing attributes from the tuples that contain a code object, removing attributes from the tuples that contain a link to a code object, etc.). In some embodiments, the TIM360may disable smart stream operation by no longer providing the segments of code wrapped in the smart tuples370with access to processing cycles of the compute nodes310. In some embodiments, the TIM360may disable smart stream operation by no longer providing access to processing cycles of additional computing systems (not depicted).

FIG. 4depicts an example method400for a smart tuple to conditionally perform an operation consistent with embodiments of the present disclosure. Method400may be executed by one or more smart tuples in a smart stream application to perform operations in response to the presence of a condition (e.g., activating a trigger in response to a condition). Method400may perform operations in addition to those depicted inFIG. 4. Method400may be executed by a given smart tuple during access to processing cycles from a compute node or other computer system. Thus, the stream application or one or more compute nodes providing processor, memory, and input/output to the stream application may also be executing method400.

From start402, a smart tuple may retrieve a relevant value at410. The smart tuple may retrieve a relevant value at410by utilizing a network of the smart stream application. The value may relate to the smart tuple (e.g., a value contained in one of the attributes of the smart tuple, the position of the smart tuple within the smart stream application, etc.) The value may relate to one or more other tuples (e.g., a value contained in an attribute of another tuple, the presence of a tuple, the location of a tuple, etc.). The value may relate to a component of the stream application (e.g., a value of a management system, a value of a source, the status of a stream operator, the status of a processing element, the status of a network utilized by the stream application, etc.). The value may relate to information outside of the stream application (e.g., retrieving a temperature value from a website on the Internet). For example, the status of a stream operator may relate to its availability, such as the latency in response time from a ping request. In a second example, the status of a processing element may relate to the resources a compute node allocates during the processing of tuples (e.g., the number of processing cycles used by a processing element, the kilobytes of memory used by a processing element, etc.).

The smart tuple may determine if the value meets a condition at420. The condition may include one or more criteria. The determination of the condition may be done by comparing the value to a single criterion, such as verifying if an attribute of the smart tuple is equal to ‘42’. The determination of the condition may be done by comparing the value to a range of values, such as verifying if an attribute is greater than ‘20’ and less than ‘32.5’. The determination of the condition may be done by verifying that a value is present (e.g., ensuring a value from a processing element is not null). The condition may be that a particular stream operator responds to a ping. In some embodiments, the determination of the condition may be done by comparing multiple values to multiple criteria (e.g., verifying a first value is equal to ‘12.7’ and a second value is not null).

If the condition is present, at430, the smart tuple may perform an operation at435. In embodiments where the condition includes multiple criteria, the condition may not be present, at430, unless all criteria are met. The operation may be activating a trigger, such as generating a notification. The operation may be setting a value such as changing an attribute of the smart tuple. The operation may be communicating a message to a management system of the smart stream application. If the condition is not present, at430, method400may end at450.

For example, a smart tuple executing method400may check an operation of a stream operator. The stream operator may be in a stalled state and unresponsive to the administrative systems of the stream application. The smart tuple may obtain an operating status of the stream operator and determine from the operating status a network connectivity issue. The smart tuple may determine that the condition of the stream operator is that it is processing tuples but has lost connectivity to parts of the network that connect it to the administrative systems of the stream application. Based on the network connectivity issue the stream operator may transmit a copy of the operating status and an identifier of the stream operator to one or more of the administrative systems of the stream application.

FIG. 5depicts the representative major components of an example computer system501that may be used, in accordance with embodiments of the present disclosure. It is appreciated that individual components may vary in complexity, number, type, and\or configuration. The particular examples disclosed are for example purposes only and are not necessarily the only such variations. The computer system501may comprise a processor510, memory520, an input/output interface (herein I/O or I/O interface)530, and a main bus540. The main bus540may provide communication pathways for the other components of the computer system501. In some embodiments, the main bus540may connect to other components such as a specialized digital signal processor (not depicted).

The processor510of the computer system501may be comprised of one or more cores512A,512B,512C,512D (collectively512). The processor510may additionally include one or more memory buffers or caches (not depicted) that provide temporary storage of instructions and data for the cores512. The cores512may perform instructions on input provided from the caches or from the memory520and output the result to caches or the memory. The cores512may be comprised of one or more circuits configured to perform one or methods consistent with embodiments of the present disclosure. In some embodiments, the computer system501may contain multiple processors510. In some embodiments, the computer system501may be a single processor510with a singular core512.

The memory520of the computer system501may include a memory controller522. In some embodiments, the memory520may comprise a random-access semiconductor memory, storage device, or storage medium (either volatile or non-volatile) for storing data and programs. In some embodiments, the memory may be in the form of modules (e.g., dual in-line memory modules). The memory controller522may communicate with the processor510, facilitating storage and retrieval of information in the memory520. The memory controller522may communicate with the I/O interface530, facilitating storage and retrieval of input or output in the memory520.

The I/O interface530may comprise an I/O bus550, a terminal interface552, a storage interface554, an I/O device interface556, and a network interface558. The I/O interface530may connect the main bus540to the I/O bus550. The I/O interface530may direct instructions and data from the processor510and memory520to the various interfaces of the I/O bus550. The I/O interface530may also direct instructions and data from the various interfaces of the I/O bus550to the processor510and memory520. The various interfaces may include the terminal interface552, the storage interface554, the I/O device interface556, and the network interface558. In some embodiments, the various interfaces may include a subset of the aforementioned interfaces (e.g., an embedded computer system in an industrial application may not include the terminal interface552and the storage interface554).

Logic modules throughout the computer system501—including but not limited to the memory520, the processor510, and the I/O interface530—may communicate failures and changes to one or more components to a hypervisor or operating system (not depicted). The hypervisor or the operating system may allocate the various resources available in the computer system501and track the location of data in memory520and of processes assigned to various cores512. In embodiments that combine or rearrange elements, aspects and capabilities of the logic modules may be combined or redistributed. These variations would be apparent to one skilled in the art.