3. 使用脉冲进行异步通知

任务目标

本案例实现一个基于QNX消息传递的脉冲（Pulse）异步通知机制：服务进程创建通道并等待客户端连接，客户端注册后，服务进程通过MsgSendPulse()向客户端发送脉冲，客户端在MsgReceive()中接收并处理该脉冲。通过本案例理解脉冲作为轻量级异步通知手段的用途——无需回复、不阻塞发送方、用于事件通知或中断延迟处理。

核心概念

脉冲是QNX中一种特殊消息：长度为0，不期望回复，发送方调用后立即返回。它常用于中断服务程序（ISR）向线程发通知、服务器向客户端推送事件等场景。接收方通过MsgReceive()获取脉冲时，rcvid为0，消息头中的type字段区分脉冲类型（系统保留_PULSE_CODE_MINAVAIL以下值，用户自定义值需大于等于该常量）。

完整实现代码

// pulse_demo.c
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <sys/neutrino.h>
#include <sys/types.h>
#include <process.h>
#include <stdlib.h>
#include <unistd.h>

#define PULSE_CODE_NOTIFY  1   // 自定义脉冲编码（需 >= _PULSE_CODE_MINAVAIL）

typedef struct {
    int client_pid;
    int client_chid;
} client_info_t;

// 服务进程：创建通道，等待连接，收到客户端注册后发送脉冲
void server_process(void) {
    int chid = ChannelCreate(0);
    if (chid == -1) {
        perror("ChannelCreate");
        exit(EXIT_FAILURE);
    }
    printf("[Server] Channel created, chid=%d\n", chid);
    fflush(stdout);

    // 通知客户端通道已就绪（通过共享文件方式传递chid）
    FILE *fp = fopen("/tmp/server_chid.txt", "w");
    if (fp) {
        fprintf(fp, "%d", chid);
        fclose(fp);
    }

    client_info_t client = {0, 0};
    int rcvid;

    while (1) {
        rcvid = MsgReceive(chid, &client, sizeof(client), NULL);
        if (rcvid == -1) {
            perror("MsgReceive");
            continue;
        }

        if (rcvid == 0) {
            // 收到脉冲（本服务端不预期收到脉冲，此处仅作完整性处理）
            printf("[Server] Received pulse\n");
            continue;
        }

        // 收到客户端注册消息（普通消息）
        printf("[Server] Client registered: pid=%d, chid=%d\n",
               client.client_pid, client.client_chid);
        fflush(stdout);

        // 向客户端发送脉冲（异步通知）
        int status = MsgSendPulse(client.client_pid, client.client_chid,
                                  PULSE_CODE_NOTIFY, 0);
        if (status == -1) {
            printf("[Server] MsgSendPulse failed: %s\n", strerror(errno));
        } else {
            printf("[Server] Pulse sent to client (pid=%d, chid=%d)\n",
                   client.client_pid, client.client_chid);
            fflush(stdout);
        }

        // 回复客户端注册消息（告知已收到）
        MsgReply(rcvid, EOK, NULL, 0);
    }
}

// 客户端进程：连接服务端，注册自身通道，等待脉冲
void client_process(void) {
    // 读取服务端chid
    FILE *fp = fopen("/tmp/server_chid.txt", "r");
    if (!fp) {
        perror("fopen server_chid");
        exit(EXIT_FAILURE);
    }
    int server_chid;
    fscanf(fp, "%d", &server_chid);
    fclose(fp);

    // 创建自身通道用于接收脉冲
    int chid = ChannelCreate(0);
    if (chid == -1) {
        perror("ChannelCreate");
        exit(EXIT_FAILURE);
    }
    printf("[Client] Channel created, chid=%d\n", chid);
    fflush(stdout);

    // 连接服务端
    int coid = ConnectAttach(0, getppid(), server_chid, 0, 0);
    if (coid == -1) {
        perror("ConnectAttach");
        exit(EXIT_FAILURE);
    }

    // 发送注册消息（包含自身pid和chid）
    client_info_t info;
    info.client_pid = getpid();
    info.client_chid = chid;
    if (MsgSend(coid, &info, sizeof(info), NULL, 0) == -1) {
        perror("MsgSend");
        exit(EXIT_FAILURE);
    }
    printf("[Client] Registration sent\n");
    fflush(stdout);

    // 等待接收脉冲
    struct _pulse pulse;
    int rcvid = MsgReceive(chid, &pulse, sizeof(pulse), NULL);
    if (rcvid == 0) {
        if (pulse.code == PULSE_CODE_NOTIFY) {
            printf("[Client] Received pulse: code=%d, value=%d\n",
                   pulse.code, pulse.value.sival_int);
            printf("[Client] Async notification received successfully!\n");
        } else {
            printf("[Client] Received unexpected pulse code=%d\n", pulse.code);
        }
    } else {
        printf("[Client] MsgReceive returned rcvid=%d (expected 0 for pulse)\n", rcvid);
    }

    ConnectDetach(coid);
    ChannelDestroy(chid);
    exit(EXIT_SUCCESS);
}

int main(int argc, char *argv[]) {
    if (argc > 1 && strcmp(argv[1], "server") == 0) {
        server_process();
    } else if (argc > 1 && strcmp(argv[1], "client") == 0) {
        client_process();
    } else {
        printf("Usage: %s [server|client]\n", argv[0]);
        return EXIT_FAILURE;
    }
    return EXIT_SUCCESS;
}

编译与运行

# 编译（QNX Momentics或qcc）
qcc -o pulse_demo pulse_demo.c

# 终端1：启动服务进程
./pulse_demo server

# 终端2：启动客户端进程（需在服务启动后执行）
./pulse_demo client

验证方法与通过标准

  1. 服务端输出：应依次显示Channel created、Client registered: pid=..., chid=...、Pulse sent to client。
  2. 客户端输出：应显示Channel created、Registration sent、Received pulse: code=1, value=0以及Async notification received successfully!。
  3. 判定通过：客户端成功打印“Received pulse”即证明服务端通过MsgSendPulse()发送的异步通知已被客户端正确接收，脉冲机制工作正常。

关键说明

  • 客户端通过ConnectAttach(0, getppid(), server_chid, ...)连接服务端，其中getppid()获取服务进程PID（因客户端由服务端fork/exec启动时父进程为服务端）。若独立启动两个进程，需将getppid()替换为服务端实际PID。
  • 服务端通过临时文件/tmp/server_chid.txt传递通道ID，这是进程间通信的简化方式；实际工程中可使用命令行参数或环境变量。
  • 脉冲接收时MsgReceive()返回的rcvid为0，消息内容为struct _pulse，其中code字段区分脉冲类型，value.sival_int可携带附加数据。
  • 本案例中服务端发送脉冲后立即继续运行，无需等待客户端处理，体现了脉冲“发送后即返回”的异步特性。
← 上一章 📖 返回目录 下一章 →