Morpheus Control Messages

(Latest Version)

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.

Copy
Copied!
            

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.

Copy
Copied!
            

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.

Copy
Copied!
            

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.

Copy
Copied!
            

import cudf from morpheus.messages import MultiMessage, ControlMessage data = cudf.DataFrame() msg_meta = MessageMeta(data)

Functionality

MultiMessage

ControlMessage

Initialization multi_msg = MultiMessage(msg_meta) control_msg = ControlMessage()
control_msg.payload(msg_meta)
Get columns from cudf.DataFrame multi_msg.get_meta(col_name) control_msg.payload().get_data(col_name)
Set columns values to cudf.DataFrame multi_msg.set_meta(col_name, value) control_msg.payload().set_data(col_name, value)
Get sliced cudf.DataFrame for given start and stop positions multi_msg.get_slice(start, stop) control_msg.payload().get_slice(start, stop)
Copy the cudf.DataFrame for given ranges of rows multi_msg.copy_ranges(ranges) control_msg.payload().copy_ranges(ranges)

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.

Previous C++ Morpheus Modules
Next Introduction to Modular Digital Fingerprinting Pipeline (Integrated Training)
© Copyright 2024, NVIDIA. Last updated on Jul 8, 2024.