【实时Linux工业PLC解决方案系列】第四篇 - 实时Linux PLC runtime核心模块开发

一、简介：为什么需要自研 PLC Runtime？

• 市场痛点：西门子、三菱等国外 PLC 价格高昂，授权绑定，二次开发受限；国产替代需求迫切但技术门槛高。

• 实时 Linux 优势：开源生态丰富、硬件支持广泛、可深度定制，是构建自主可控 PLC 的理想底座。

• Runtime 核心价值：作为 PLC 系统的"心脏"，负责任务调度、I/O 处理、程序解析，直接决定控制精度与稳定性。

• 掌握收益：具备自主 PLC 平台能力，可快速响应定制化需求（如特殊通信协议、行业算法库），打破国外垄断。

二、核心概念：PLC Runtime 架构与术语

三、环境准备：搭建开发环境

3.1 硬件需求

3.2 软件环境

3.3 一键编译实时内核

#!/bin/bash # build_rt_kernel.sh set -e VERSION=5.15.71 RT_PATCH=patch-5.15.71-rt53.patch.xz wget https://cdn.kernel.org/pub/linux/kernel/v5.x/linux-${VERSION}.tar.xz wget https://cdn.kernel.org/pub/linux/kernel/projects/rt/${VERSION}/${RT_PATCH} tar -xf linux-${VERSION}.tar.xz cd linux-${VERSION} xzcat ../${RT_PATCH} | patch -p1 # 配置：启用 PREEMPT_RT，关闭调试 make x86_64_defconfig ./scripts/config --enable CONFIG_PREEMPT_RT ./scripts/config --disable CONFIG_DEBUG_PREEMPT ./scripts/config --disable CONFIG_DYNAMIC_DEBUG make -j$(nproc) deb-pkg sudo dpkg -i ../linux-image-${VERSION}*.deb sudo dpkg -i ../linux-headers-${VERSION}*.deb echo "重启选择 RT 内核"

四、应用场景：智能产线多轴同步控制

在新能源汽车电池模组装配线上，6 台 SCARA 机械臂需以 1ms 周期同步完成电芯抓取、视觉定位、激光焊接动作。传统 PLC 因扫描周期固定为 10ms，无法满足 1ms 级多轴插补需求。采用实时 Linux PLC 方案后：

• 周期任务：1ms 级运动控制算法（SFC 实现状态机）

• 事件任务：视觉检测结果到达时触发轨迹修正（ST 实现插值计算）

• 中断任务：急停信号 50μs 内响应（LD 实现安全逻辑）

通过自研 Runtime 的任务调度器，实现三种任务的无冲突调度，轴间同步误差 < 0.1mm，焊接良品率从 92% 提升至 99.5%。

五、实际案例与步骤：Runtime 核心模块开发

5.1 项目目录结构

mkdir -p plc_runtime/{scheduler,io_manager,parser,common,tests} cd plc_runtime touch CMakeLists.txt

5.2 任务调度器（Scheduler）

设计目标：支持周期/事件/中断三级任务，抖动 < 50μs。

/* scheduler/scheduler.h */ #ifndef SCHEDULER_H #define SCHEDULER_H #include <stdint.h> #include <pthread.h> #include <time.h> /* 任务类型 */ typedef enum { TASK_CYCLIC, /* 周期任务：固定间隔执行 */ TASK_EVENT, /* 事件任务：条件触发 */ TASK_INTERRUPT /* 中断任务：最高优先级，紧急响应 */ } task_type_t; /* 任务优先级（1-99，SCHED_FIFO） */ #define PRIO_CYCLIC 50 #define PRIO_EVENT 70 #define PRIO_INTERRUPT 99 /* 任务控制块 */ typedef struct { uint32_t id; task_type_t type; uint32_t period_us; /* 周期任务：微秒 */ void (*func)(void*); /* 任务函数指针 */ void* arg; pthread_t thread; struct timespec next_run; volatile int triggered; /* 事件/中断触发标志 */ } task_t; /* 调度器接口 */ int scheduler_init(void); int task_create(uint32_t id, task_type_t type, uint32_t period_us, void (*func)(void*), void* arg); int task_start(uint32_t id); int task_trigger(uint32_t id); /* 触发事件/中断任务 */ void scheduler_run(void); /* 主调度循环 */ #endif

/* scheduler/scheduler.c */ #define _GNU_SOURCE #include "scheduler.h" #include <stdio.h> #include <stdlib.h> #include <string.h> #include <sched.h> #include <errno.h> #define MAX_TASKS 32 static task_t* tasks[MAX_TASKS]; static int task_count = 0; /* 设置实时优先级 */ static int set_rt_priority(pthread_t thread, int prio) { struct sched_param param = { .sched_priority = prio }; return pthread_setschedparam(thread, SCHED_FIFO, &param); } int scheduler_init(void) { memset(tasks, 0, sizeof(tasks)); return 0; } int task_create(uint32_t id, task_type_t type, uint32_t period_us, void (*func)(void*), void* arg) { if (task_count >= MAX_TASKS) return -1; task_t* t = calloc(1, sizeof(task_t)); t->id = id; t->type = type; t->period_us = period_us; t->func = func; t->arg = arg; t->triggered = 0; tasks[task_count++] = t; return 0; } /* 周期任务线程 */ static void* cyclic_thread(void* arg) { task_t* t = arg; struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); while (1) { t->func(t->arg); /* 执行用户程序 */ /* 计算下一次执行时间 */ ts.tv_nsec += t->period_us * 1000; if (ts.tv_nsec >= 1000000000) { ts.tv_sec++; ts.tv_nsec -= 1000000000; } clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &ts, NULL); } return NULL; } /* 事件/中断任务线程 */ static void* event_thread(void* arg) { task_t* t = arg; while (1) { while (!t->triggered) { __sync_synchronize(); /* 内存屏障 */ } t->triggered = 0; t->func(t->arg); /* 执行用户程序 */ } return NULL; } int task_start(uint32_t id) { for (int i = 0; i < task_count; i++) { if (tasks[i]->id != id) continue; task_t* t = tasks[i]; int prio = (t->type == TASK_CYCLIC) ? PRIO_CYCLIC : (t->type == TASK_EVENT) ? PRIO_EVENT : PRIO_INTERRUPT; pthread_attr_t attr; pthread_attr_init(&attr); pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED); pthread_attr_setschedpolicy(&attr, SCHED_FIFO); struct sched_param param = { .sched_priority = prio }; pthread_attr_setschedparam(&attr, &param); void* (*thread_func)(void*) = (t->type == TASK_CYCLIC) ? cyclic_thread : event_thread; pthread_create(&t->thread, &attr, thread_func, t); return 0; } return -1; } int task_trigger(uint32_t id) { for (int i = 0; i < task_count; i++) { if (tasks[i]->id == id) { tasks[i]->triggered = 1; __sync_synchronize(); return 0; } } return -1; }

编译测试：

gcc -o test_scheduler scheduler/scheduler.c -pthread -lrt sudo ./test_scheduler # 需 root 设置实时优先级

5.3 I/O 映像区管理（I/O Manager）

设计目标：硬件 I/O 与程序解耦，支持热插拔，读写延迟 < 10μs。

/* io_manager/io_manager.c */ #include "io_manager.h" #include <string.h> #include <time.h> static io_point_t pi_image[PI_SIZE]; /* 过程输入 */ static io_point_t pq_image[PQ_SIZE]; /* 过程输出 */ static pthread_rwlock_t pi_lock; /* 读写锁：读者优先 */ static pthread_rwlock_t pq_lock; int io_manager_init(void) { memset(pi_image, 0, sizeof(pi_image)); memset(pq_image, 0, sizeof(pq_image)); pthread_rwlock_init(&pi_lock, NULL); pthread_rwlock_init(&pq_lock, NULL); return 0; } /* 程序读取数字输入 */ bool io_read_di(uint32_t addr) { if (addr >= PI_SIZE) return false; pthread_rwlock_rdlock(&pi_lock); bool val = (pi_image[addr].type == IO_DI) ? pi_image[addr].value.digital : false; pthread_rwlock_unlock(&pi_lock); return val; } /* 程序写入数字输出 */ void io_write_do(uint32_t addr, bool value) { if (addr >= PQ_SIZE) return; pthread_rwlock_wrlock(&pq_lock); pq_image[addr].type = IO_DO; pq_image[addr].value.digital = value; pq_image[addr].timestamp = clock_gettime_ns(); pthread_rwlock_unlock(&pq_lock); } /* 硬件同步：EtherCAT 示例 */ void io_sync_input(void) { /* 从 EtherCAT 从站读取 */ ec_slave_config_t* slave = ecrt_master_slave_config(...); uint8_t* input_data = ecrt_slave_config_input_data(slave); pthread_rwlock_wrlock(&pi_lock); for (int i = 0; i < PI_SIZE && i < slave->input_size; i++) { pi_image[i].value.digital = (input_data[i/8] >> (i%8)) & 1; pi_image[i].timestamp = clock_gettime_ns(); } pthread_rwlock_unlock(&pi_lock); }

5.4 编程语言解析器（Parser）

设计目标：支持 LD/FBD/ST/SFC 四种语言，统一编译为字节码。

/* parser/ast.h */ #ifndef AST_H #define AST_H /* 抽象语法树节点类型 */ typedef enum { AST_LD_CONTACT, /* LD: 常开触点 */ AST_LD_COIL, /* LD: 线圈 */ AST_FBD_BLOCK, /* FBD: 功能块 */ AST_ST_ASSIGN, /* ST: 赋值 */ AST_ST_IF, /* ST: 条件 */ AST_SFC_STEP, /* SFC: 步 */ AST_SFC_TRANSITION /* SFC: 转换条件 */ } ast_node_type_t; typedef struct ast_node { ast_node_type_t type; union { struct { char var[32]; bool normally_open; } contact; struct { char var[32]; } coil; struct { char name[32]; struct ast_node** inputs; } block; struct { char target[32]; struct ast_node* expr; } assign; struct { struct ast_node* cond; struct ast_node* then_stmt; struct ast_node* else_stmt; } if_stmt; } data; struct ast_node* next; /* 链表连接 */ } ast_node_t; /* 字节码指令 */ typedef enum { OP_LOAD, OP_STORE, OP_AND, OP_OR, OP_NOT, OP_ADD, OP_SUB, OP_MUL, OP_DIV, OP_JMP, OP_JZ, OP_CALL, OP_RET } opcode_t; typedef struct { opcode_t op; uint32_t operand; } instruction_t; /* 解析接口 */ ast_node_t* parse_ld(const char* source); /* 梯形图 */ ast_node_t* parse_st(const char* source); /* 结构化文本 */ instruction_t* compile_ast(ast_node_t* root); /* AST → 字节码 */ #endif

/* parser/st_parser.c - ST 子集示例 */ #include "ast.h" #include <stdio.h> #include <stdlib.h> #include <string.h> #include <ctype.h> static const char* input; static int pos; static void skip_space() { while (isspace(input[pos])) pos++; } static char* parse_identifier() { skip_space(); char buf[32]; int i = 0; while (isalnum(input[pos]) && i < 31) { buf[i++] = input[pos++]; } buf[i] = '\0'; return strdup(buf); } /* ST: VAR := EXPR; */ ast_node_t* parse_assignment() { ast_node_t* node = calloc(1, sizeof(ast_node_t)); node->type = AST_ST_ASSIGN; char* var = parse_identifier(); strcpy(node->data.assign.target, var); free(var); skip_space(); if (input[pos] == ':' && input[pos+1] == '=') { pos += 2; /* 简化：只解析数字常量 */ skip_space(); int value = 0; while (isdigit(input[pos])) { value = value * 10 + (input[pos++] - '0'); } /* 创建常量表达式节点 */ ast_node_t* expr = calloc(1, sizeof(ast_node_t)); expr->type = AST_FBD_BLOCK; /* 复用作为常量 */ sprintf(expr->data.block.name, "%d", value); node->data.assign.expr = expr; } skip_space(); if (input[pos] == ';') pos++; return node; } /* 编译为字节码 */ instruction_t* compile_ast(ast_node_t* root) { static instruction_t code[256]; int pc = 0; for (ast_node_t* n = root; n; n = n->next) { switch (n->type) { case AST_ST_ASSIGN: /* 生成表达式代码 */ code[pc++] = (instruction_t){ OP_LOAD, atoi(n->data.assign.expr->data.block.name) }; code[pc++] = (instruction_t){ OP_STORE, hash(n->data.assign.target) }; break; /* ... 其他类型 */ } } code[pc++] = (instruction_t){ OP_RET, 0 }; return code; }

5.5 字节码解释器（Executor）

/* common/executor.c */ #include "ast.h" #include <stdint.h> #define STACK_SIZE 256 static int32_t stack[STACK_SIZE]; static int sp = 0; #define VAR_TABLE_SIZE 1024 static int32_t variables[VAR_TABLE_SIZE]; void execute(instruction_t* code, io_manager_t* io) { int pc = 0; while (1) { instruction_t inst = code[pc++]; switch (inst.op) { case OP_LOAD: stack[sp++] = inst.operand; break; case OP_STORE: if (inst.operand < VAR_TABLE_SIZE) variables[inst.operand] = stack[--sp]; break; case OP_AND: stack[sp-2] = stack[sp-2] & stack[sp-1]; sp--; break; case OP_OR: stack[sp-2] = stack[sp-2] | stack[sp-1]; sp--; break; case OP_JZ: if (stack[--sp] == 0) pc = inst.operand; break; case OP_CALL: /* 调用系统功能块 */ break; case OP_RET: return; } } }

5.6 主程序集成

/* main.c */ #include "scheduler/scheduler.h" #include "io_manager/io_manager.h" #include "parser/ast.h" #include <stdio.h> /* 用户程序示例：1ms 周期任务 */ void motor_control(void* arg) { static int cycle = 0; /* 读取编码器 */ int32_t pos = io_read_ai(0); /* PID 计算（简化） */ int32_t target = 1000; int32_t error = target - pos; int32_t output = error * 2; /* Kp=2 */ /* 输出到驱动器 */ io_write_ao(0, output); if (++cycle % 1000 == 0) printf("Motor control: 1000 cycles completed\n"); } /* 事件任务：急停处理 */ void emergency_stop(void* arg) { printf("EMERGENCY STOP triggered!\n"); io_write_do(0, false); /* 切断主接触器 */ /* 进入安全状态 */ } int main() { /* 初始化 */ scheduler_init(); io_manager_init(); /* 注册 I/O */ io_register(0, IO_AI); /* 编码器 */ io_register(0, IO_AO); /* 驱动器 */ io_register(0, IO_DO); /* 接触器 */ /* 创建任务 */ task_create(1, TASK_CYCLIC, 1000, motor_control, NULL); /* 1ms */ task_create(2, TASK_INTERRUPT, 0, emergency_stop, NULL); /* 中断 */ /* 启动 */ task_start(1); task_start(2); /* 模拟外部中断触发 */ sleep(5); task_trigger(2); /* 主循环：I/O 同步 */ while (1) { io_sync_input(); /* 调度器在各自线程运行 */ io_sync_output(); usleep(100); /* 100μs 微调 */ } return 0; }

CMake 构建：

# CMakeLists.txt cmake_minimum_required(VERSION 3.16) project(plc_runtime) set(CMAKE_C_STANDARD 11) set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -O2 -Wall -Wextra") add_executable(plc_runtime main.c scheduler/scheduler.c io_manager/io_manager.c parser/st_parser.c common/executor.c ) target_link_libraries(plc_runtime pthread rt) # 测试 enable_testing() add_subdirectory(tests)

六、常见问题与解答

七、实践建议与最佳实践

1. 调度器调优

   • 周期任务绑定独立 CPU 核心：taskset -c 2 ./plc_runtime

   • 使用SCHED_DEADLINE替代SCHED_FIFO，确保最坏执行时间约束

2. I/O 性能

   • 采用双缓冲：硬件线程写 Buffer A，程序读 Buffer B，周期结束时原子交换

   • DMA 传输 > MMIO > 端口 I/O，优先使用支持 DMA 的 I/O 板卡

3. 语言解析

   • 使用flex/bison生成完整解析器，替代手写递归下降

   • 引入 LLVM 后端，将字节码编译为机器码，提升 5-10 倍执行效率

4. 调试技巧

   • 内置示波器功能：将关键变量通过 EtherCAT 上传，上位机实时绘制波形

   • 使用ftrace跟踪调度延迟：trace-cmd record -e sched_switch

5. 安全设计

   • 看门狗线程：监控周期任务执行，超时 2 个周期即复位

   • 安全 PLC 分区：关键安全功能运行在独立核，与非安全应用隔离

八、总结与应用场景

本文系统讲解了实时 Linux PLC runtime 的核心模块开发，包括：

掌握这些技能，你可以：

• 替代进口 PLC，成本降低 60% 以上

• 定制专用协议（如自有伺服驱动器通信）

• 集成 AI 算法（在周期任务中调用 TensorRT 推理）