Morpheus Control Messages

Control Messages, introduced in version 23.03, provide a solution for numerous use cases that were previously unattainable. This new paradigm enhances the capabilities of Morpheus pipelines by enabling more reactive, event-driven operations. Control Messages involve sending message objects to a pipeline, which can represent a wide range of concepts, from raw data to explicit directives for loading data from specified sources or initiating out-of-band inference or training tasks. The pipeline’s behavior can dynamically adapt based on the design; some stages may disregard messages they are not intended to process, while others act according to the message type and content.

This approach unlocks various new applications for Morpheus pipelines. For instance, Control Messages can facilitate real-time data processing and analysis, allowing pipelines to respond promptly to time-sensitive events or data streams. Additionally, they can support adaptive machine learning models that continuously update and refine their predictions based on incoming data. Furthermore, Control Messages can improve resource allocation and efficiency by enabling on-demand data processing and task execution. Overall, the introduction of Control Messages in Morpheus pipelines paves the way for more versatile and responsive software solutions, catering to a broader range of requirements and use cases.

Control Messages are straightforward objects that contain tasks, metadata, and possibly payload data. Tasks can be one of the following: training, inference, or other. Metadata is a dictionary of key-value pairs that provide additional information about the message and must be JSON serializable. Payload is a Morpheus MessageMeta object that can be used to move raw data. Each of these elements can be accessed via the API as the message flows through the pipeline.

Working with Tasks

Control Messages can handle tasks such as training, inference, and a catchall category other. Tasks can be added, checked for existence, or removed from the Control Message using methods like add_task, has_task, and remove_task.

            

            
import morpheus._lib.messages as messages

task_data = {
    "....": "...."
}

msg = messages.ControlMessage()
msg.add_task("training", task_data)
if msg.has_task("training"):
    task = msg.remove_task("training")
        

Managing Metadata

Metadata is a set of key-value pairs that offer supplementary information about the Control Message and must be JSON serializable. You can set, check, and retrieve metadata values using the set_metadata, has_metadata, and get_metadata methods, respectively.

            

            
import morpheus._lib.messages as messages

msg = messages.ControlMessage()
msg.set_metadata("description", "This is a sample control message.")
if msg.has_metadata("description"):
    description = msg.get_metadata("description")
        

Handling Payloads

The payload of a Control Message is a Morpheus MessageMeta object that can carry raw data. You can set or retrieve the payload using the payload method, which can accept a MessageMeta instance or return the payload itself.

            

            
import cudf
import morpheus._lib.messages as messages

data = cudf.DataFrame()  # some data

msg_meta = messages.MessageMeta(data)
msg = messages.ControlMessage()

msg.payload(msg_meta)

retrieved_payload = msg.payload()

msg_meta == retrieved_payload # True
        

Conversion from MultiMessage to ControlMessage

Starting with version 24.06, the MultiMessage type will be deprecated, and all usage should transition to ControlMessage. Each MultiMessage functionality has a corresponding equivalent in ControlMessage, as illustrated below.

            

            
import cudf
from morpheus.messages import MultiMessage, ControlMessage

data = cudf.DataFrame()
msg_meta = MessageMeta(data)
        

Note that the get_slice() and copy_ranges() functions in ControlMessage return the MessageMeta after slicing, whereas these functions in MultiMessage return a new MultiMessage instance.