Dispatching an Operator

After orchestrating tasks for a communication operator, you need to dispatch the kernel function to the specific communication engine for execution.

For the AICPU communication engine, you need to declare the operator information library file in JSON format. The file content is as follows:

{
  "CustomAicpuKernel": {
    "opInfo": {
      "opKernelLib": "AICPUKernel",                // Fixed value
      "kernelSo": "<aicpu_kernel_so_name>",        // Dynamic link library on the AICPU, which is user-defined, for example, libp2p_aicpu_kernel.so
      "functionName": "<aicpu_kernel_func_name>"   // AICPU kernel function name, which is user-defined, for example, HcclLaunchP2PAicpuKernel
    }
  }
}

Define the AICPU kernel function in the following format. <aicpu_kernel_func_name> indicates the AICPU kernel function name, which must be the same as the value of the functionName field in the operator information library file (JSON file).

extern "C" unsigned int <aicpu_kernel_func_name>(void *param)
{
    // Kernel implementation
}

Sample Code

The following uses the custom Send/Receive operators as an example to describe the code snippet for dispatching the AICPU kernel function on the host when the AICPU communication engine is used:

// The host stream notifies the device of the primary thread.
aclrtRecordNotify(g_notifies[0], stream);

std::string kernelName = "HcclLaunchP2PAicpuKernel";
aclrtFuncHandle funcHandle;
aclrtArgsHandle argsHandle;
ACLCHECK(aclrtBinaryGetFunction(g_binKernelHandle, kernelName.c_str(), &funcHandle));
ACLCHECK(aclrtKernelArgsInit(funcHandle, &argsHandle));
aclrtParamHandle paraHandle;
aclrtKernelArgsAppend(argsHandle, &param, sizeof(OpParam), &paraHandle);
aclrtKernelArgsFinalize(argsHandle);

uint16_t NOTIFY_DEFAULT_WAIT_TIME = 27 * 68;
aclrtLaunchKernelCfg cfg;
aclrtLaunchKernelAttr attr;
attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
attr.value.timeout = NOTIFY_DEFAULT_WAIT_TIME;
cfg.numAttrs = 1;
cfg.attrs = &attr;
constexpr uint32_t blockDim = 1;

// Orchestrate algorithms on the device.
aclrtLaunchKernelWithConfig(funcHandle, blockDim, stream, &cfg, argsHandle, nullptr);

// The host stream waits for the notification from the device.
aclrtWaitAndResetNotify(g_notifies[1], stream, CUSTOM_TIMEOUT);

Define the kernel function entry on the AICPU. The function needs to be compiled on the device.

typedef struct {
    void *addr;
    uint64_t size;
} CommBuffer;

struct AlgResourceCtx {
    ThreadHandle threadHandle;                              // Communication thread handle
    CommBuffer localBuffer;                                 // Local communication memory
    CommBuffer remoteBuffer;                                // Remote communication memory
    ChannelHandle channelHandle;                            // Communication channel resource
    uint32_t notifyIds[AICPU_CONTROL_NOTIFY_NUM];           // Control Notify on the device in AICPU mode
};

struct OpParam {
    char tag[TAG_LENGTH];
    char commName[COMM_INDENTIFIER_MAX_LENGTH];             // Communicator name
    void* inputPtr = nullptr;                               // Operator input data address
    void* outputPtr = nullptr;                              // Operator output data address
    uint64_t count = 0;                                     // Operator data size
    HcclDataType dataType = HCCL_DATA_TYPE_RESERVED;        // Operator data type
    HcclCMDType opType = HcclCMDType::HCCL_CMD_INVALID;     // Operator type
    AlgResourceCtx* resCtx = nullptr;                       // Resource context
};

// Kernel function executed on the AICPU
extern "C" unsigned int HcclLaunchP2PAicpuKernel(OpParam *param)
{
    HCCL_INFO("Entry-%s, commName[%s], tag[%s]", __func__, param->commName, param->tag);
    if (HcommAcquireComm(param->commName) != HCCL_SUCCESS) { 
        HCCL_ERROR("%s HcommAcquireComm fail, commName[%s]", __func__, param->commName);
        return 1;
    }

    //Obtain the primary thread on the device.
    ThreadHandle thread = param->resCtx->threadHandle;
    if (HcommBatchModeStart(param->tag) != HCCL_SUCCESS) {
        HCCL_ERROR("failed start batch mode");
        return 1;
    }

    // The primary thread waits for the notification from the host stream.
    if (HcommAclrtNotifyWaitOnThread(thread, param->resCtx->notifyIds[0], CUSTOM_TIMEOUT) != HCCL_SUCCESS) {
        HCCL_ERROR("failed to wait notify[%d] from host main stream", param->resCtx->notifyIds[0]);
        return 1;
    }

    // Execute task orchestration.
    if (ExecOp(*param, param->resCtx) != HCCL_SUCCESS) {
        HCCL_ERROR("orchestrate failed for op:%d", param->opType);
        return 1;
    }

    // The primary thread notifies the host stream.
    if (HcommAclrtNotifyRecordOnThread(thread, param->resCtx->notifyIds[1]) != HCCL_SUCCESS) {
        HCCL_ERROR("failed to record host main stream");
        return 1;
    }

    if (HcommBatchModeEnd(param->tag) != HCCL_SUCCESS) {
        HCCL_ERROR("failed end batch mode");
        return 1;
    }

    if (HcommReleaseComm(param->commName) != HCCL_SUCCESS) {
        HCCL_ERROR("%s HcommReleaseComm fail, commName[%s]", __func__, param->commName);
        return 1;
    }
    HCCL_INFO("%s success, commName[%s], tag[%s]", __func__, param->commName, param->tag);
    return 0;
}