BufferMessage (Asynchronous Inference Scenario)

Scenario Description

When robot applications use AI models to perform inference on data, the asynchronous inference mode can be used. The inference engine (NN Engine) obtains data from the input queue, executes inference, and writes the inference result to the output queue.

To provide a consistent usage form, OpenHiva can provide data to the inference engine in the form of topics, obtain the inference result in the form of subscription, invoke the processing function for post-processing, and finally publish the inference result in the form of publishing.

This section uses the NN Engine as an example to describe the process of invoking BufferMessage messages in inference scenarios.

Principles

In the inference scenario, the internal implementation principle of sending and receiving inference messages is slightly different from the topic transfer mode of "Message Subscription - Message Publishing" in BufferMessage (Basic Scenario). The main difference is that the Hiva::CreateNnEngine interface is added. You need to transfer the input and output message names of the model to this interface during initialization. The interface automatically applies for and creates a receiving queue based on the input and output messages to receive messages to be inferred (input of the NN). The interface automatically creates a publish queue to store the inference result (output of the NN). After a callback function for processing the result is registered, the callback function is invoked once the NN inference is complete. You can add logic to the callback function for post-processing.

Figure 1 shows the process of invoking key interfaces for NN engine inference.

Figure 1 Process of invoking the NN engine inference interface

Initialize resources.
Before invoking the OpenHiva interface, invoke the OpenHiva::Init interface for initialization. If the return value is 0, the initialization is successful. Otherwise, the initialization fails.

Initialization includes node name registration, thread group creation, and resource application.
- Node name registration:
  - The names of different nodes must be unique. Otherwise, the nodes that are started later will fail to be initialized.
  - If the initialization fails, the node cannot perform the publishing or subscription action.
- Thread group creation:
  The thread group parameters need to be transferred when the OpenHiva::Init interface is invoked. Create a thread group (ScheduleGroup) as required in advance. Each thread group stores information about the callback thread (for receiving data), including the thread group name (groupName) and thread group scheduling type (scheduleType).
  
  The internal processing flow of OpenHiva varies according to the value of ThreadGroup.scheduleType in the thread group.
  - UNBIND_AICPU (recommended): indicates threads in the non-deterministic domain. The thread group is not bound to cores. Each group starts a working thread waiting for the event to be activated. The operating system schedules the CPU on which the working thread runs.
  - BIND_AICPU: indicates threads in the deterministic domain, and the thread group is bound to cores. A maximum of four working threads can be started in each group (depending on the number of CPU cores). Each working thread is bound to one core. The scheduling status is determined by the event scheduling mechanism. This mechanism provides high real-time performance, avoiding time jitter caused by kernel scheduling of the operating system to a large extent, and ensuring the determinism of thread switching from sleep to running.
  - USER_DEFINED: indicates that the working thread is not started. This type of group is not scheduled by the framework and must be used together with OpenHiva::SpinOnce. When a user invokes OpenHiva::SpinOnce, the value of groupName must be the same as that of ScheduleGroup.groupName. This interface processes messages in the thread invoked by the user.
Perform inference.
- Before inference, prepare an OM model that can be directly used for inference on the Ascend AI Processors. If the network model is based on an open source framework (such as Caffe and TensorFlow), use the Ascend Tensor Compiler (ATC) tool to convert the open source framework network model into an offline model (*.om file) that adapts to the Ascend AI Processor. For details about how to use the ATC tool, see the ATC Tool User Guide.
- You must register a callback function for each topic in outTopicInfo to process messages. If you do not process a topic, you also need to register an empty callback function for the topic.
1. Invoke the CreateNnEngine interface to perform inference. During inference, the framework loads the model file specified by modelPath and subscribes to the topic specified by inTopicInfo. After the inference is complete, the framework publishes the topic specified by outTopicInfo.
2. Invoke the CreateSubscriber interface to subscribe to the inference result, that is, the topic specified by outTopicInfo. You can customize a function to process the inference result.
Release resources.
The corresponding resources need to be released before the process ends. The defined OpenHiva APIs cannot be used.
1. Releases NN engine resources. You need to invoke the DestroyNnEngine interface to release resources. The value of engineName of this interface must be the same as that of engineName of the CreateNnEngine interface.
2. Release framework resources.
  - In the exception branch, the OpenHiva::Shutdown interface needs to be invoked to release resources, including deregistering the queue ID and topic information and releasing the corresponding memory block.
  - In the main thread, the OpenHiva::WaitForShutdown interface needs to be proactively invoked to release resources to prevent the main thread from exiting in advance. When the process stops abnormally, the WaitForShutdown interface automatically invokes the Shutdown interface to clear resources.

Sample Code

#include "hiva.h"
#include "inner_topic/nnengine.h"

const std::string nnInputTopic1 = "NN1BufferMsg";            // Receive ISP data from camera 1.
const std::string nnInputTopic2 = "NN2BufferMsg";            // Receives ISP data from camera 2.
const std::string nnOutputTopic1 = "NN1OutputBufferMsg";     // Output the inference result of the ISP data of camera 1.
const std::string nnOutputTopic2 = "NN2OutputBufferMsg";     // Outputs the inference result of the ISP data of camera 2.
const uint32_t inputTopicQueSize = 10;
const uint32_t outputTopicQueSize = 10;
const uint32_t groupId = 0;
const uint32_t priority = 0;
const std::string nnEngineName = "aclEngine1";              // engineName
const std::string modelPath = "/home/aclMode";              // Indicates the path of the OM model file.
std::vector<std::shared_ptr<OpenHiva::Subscriber>> sub;

// Register the callback function to process the model inference result.
void PerceptionCallback(const OpenHiva::BufferMessage &buffMsg)
{
    void *currBlockPtr = nullptr;
    size_t dataSize = 0U;
    OpenHiva::HivaBuffer &HivaBuffer = const_cast<OpenHiva::HivaBuffer &>(buffMsg.data);
    uint32_t ret = HivaBuffer.GetBuff(currBlockPtr, dataSize);        // Obtain the memory address dataPtr and dataLen of the data.
    if (ret != Hiva::HIVA_SUCCESS) {
        HIVA_ERROR("GetBuff failed and return %u", ret);
        return;
    }
    uint32_t dataLen = *reinterpret_cast<uint32_t *>(currBlockPtr);
    DoSthForNNaction();                                               // User-defined processing function, which is used to process the inference result.
    return;
}

// User-defined function, which is used to initialize the inference node.
uint32_t PerceptionNodeInit(OpenHiva::Node &node, const OpenHiva::TopicOptions &topicOps)
{
    uint32_t ret = HIVA_SUCCESS;
    std::vector<std::string> inputTopicVec;
    std::vector<std::string> outputTopicVec;
    inputTopicVec.push_back(nnInputTopic1);
    inputTopicVec.push_back(nnInputTopic2);
    outputTopicVec.push_back(nnOutputTopic1);
    outputTopicVec.push_back(nnOutputTopic2);
    Hiva::TopicInfo inTopicInfo = {
        .topicVec = inputTopicVec,
        .queueSize = inputTopicQueSize
    };
    Hiva::TopicInfo outTopicInfo = {
        .topicVec = outputTopicVec,
        .queueSize = outputTopicQueSize
    };
    // Create an NN engine, load the specified model file, perform inference on the data released by inTopicInfo, and publish the inference result in outTopicInfo format.
    ret = Hiva::CreateNnEngine(nnEngineName, modelPath, inTopicInfo, outTopicInfo);
    if (ret != HIVA_SUCCESS) {
        HIVA_ERROR("CreateNnEngine Fail. engineName:%s(%zu,%u,%zu,%u,%s).", nnEngineName.c_str(),
            inTopicInfo.topicVec.size(), inTopicInfo.queueSize, outTopicInfo.topicVec.size(), outTopicInfo.queueSize,
            modelPath.c_str());
        return ret;
    }
    // The value of groupName in the Subscribe interface must be the same as that of ThreadGroup.groupName in the Init interface.
    const std::string groupName = "hivatest_perception";
    for (auto itr = outputTopicVec.begin(); itr != outputTopicVec.end(); itr++) {
        // Invoke the Subscriber interface through the NodeHandle object to subscribe to the inference result topic, and invoke callback functions in sequence based on outputTopic to process the inference result. (In this example, the same callback function is used.)
        sub.push_back(node.CreateSubscriber(*itr, &PerceptionCallback,topicOps);
    }
    return HIVA_SUCCESS;
}


int main(int argc, char **argv)
{
    // 1. Initialize resources.
    std::string nodeName = "test_percetion";         
    std::string topicName = "testTopic";                   // Name of the subscribed topic, which must be the same as that of the topic published by the publisher.
    std::string groupName = "hivatest_perception";        // The value of groupName must be the same as that of ThreadGroup.groupName in the Init interface.
    OpenHiva::ScheduleType scheType = OpenHiva::ScheduleType(0);
    uint32_t maxMsgSize = 100 * 1024U;
    uint32_t queueSize = 10U;
    uint32_t blockNum = 10U;
    bool overwrite = false;
    bool queueFCFlag = false;
    uint16_t queueTTL = 100U;
    std::string transport;
    // Define a thread group.
    std::vector<OpenHiva::ScheduleGroup> scheGrpVec;
    OpenHiva::ScheduleGroup scheGrp = scheType;
    scheGrp.groupName = groupName;
    scheGrp.scheduleType = scheType;
    scheGrpVec.push_back(scheGrp);
    // Invoke the resource initialization interface.
    OpenHiva::Init(argc, argv, scheGrpVec);

    // 2. Perform inference.
    // Construct the Node object.
    OpenHiva::Node node(nodeName);
    // Construct TopicOptions.
    OpenHiva::TopicOptions topicOps;
    topicOps.BuildGroupName(groupName)
            .BuildMessageTraits<OpenHiva::BufferMessage>()
            .BuildShmOptions(maxMsgSize, blockNum)
            .BuildQueueOptions(queueSize, overwrite, queueFCFlag, queueTTL)
            .BuildTopicName(topicName);
    // Execute inference, subscribe to the inference result, and perform post-processing.
    if (PerceptionNodeInit(node, topicOps) != HIVA_SUCCESS) {
        HIVA_ERROR("PerceptionNodeInit Error.");
        OpenHiva::Shutdown();
        return HIVA_FAILED;
    }
    while (OpenHiva::Ready()) {
        HIVA_INFO("PerceptionTest is Running.");
        sleep(1);
    }

    // 3. Release resources.
    if (Hiva::DestroyNnEngine(nnEngineName) != HIVA_SUCCESS) {
        HIVA_ERROR("DestroyNnEngine failed");
    }
    OpenHiva::WaitForShutdown();
    return 0;
}

Parent topic: Message Subscribing-Publishing