目录
-
前言
-
VXCAN 打通1000条隧道
-
C 模拟1000个传感器
-
C 解析1000个传感器
-
Github
前言
得益于最近这些年高并发, 高密集IO网络的不断优化, SocketCAN也跟着沾了光. 基本上网络相关的硬件/概念/算法都可以死搬硬套到CAN上来, 本篇验证下同时处理1000个传感器的数据.
1000路CANFD设备+1000个传感器, 考虑到5Mbit/s, 至少需要万兆网卡来聚合, 加上备份, 这些设备采购下来估计几百万甚至上亿就没了. 所以本篇当然是模拟的方式:
-
VXCAN 打通1000对 canx-vcanx, can0-vxcan0, can1-vxcan1... -
用C在单线程模拟1000个传感器, 在vxcanx播发, 如果更多, 如10000个, 可能要拆成多线程让众多CPU核一块分担. (最早用Python来实现的, 发现周期达不到10ms要求, 即便开多线程对所有CPU核负载也太大) -
传感器数据解析就正常的在canx处理就好
VXCAN 打通1000条隧道
#!/bin/sh
sudo modprobe can_raw
sudo modprobe vxcan
i=0
while [ $i -le 999 ]
do
echo can$i
if ip link show can$i > /dev/null 2>&1; then
i=$(($i+1))
continue
# sudo ip link set dev can$i down
# sudo ip link set dev vxcan$i down
# sudo ip link delete dev can$i type vxcan
fi
sudo ip link add dev can$i type vxcan
sudo ip link set up can$i
sudo ip link set dev vxcan$i up
i=$(($i+1))
done
这个1000对生成的过程可能很慢, 几秒到数分钟不等. 如果某些CAN有问题, 就放出来注释掉的3行, 同时注释前面2行. 生成完后可以用ifconfig体验下刷屏的快感, 或者用cansend, candump命令随意挑选一对进行测试. 注意:
-
candump any最多同时接收30路数据, 如果想要更多, 就需要自己去拉下来can-utils的源码, 修改candump.c中MAXIFNAMES的值, 如从30改到2000或10000, 再重新make, 运行./candump any, 或者sudo make install来替换原来apt方式安装的.
挑一对来看, mtu 72, 可正常用CANFD, txqueuelen 1000, 这个默认值基本可以保证短时间内不丢帧了, 实在不够还能手动加.
$ ifconfig can888
can888: flags=193<UP,RUNNING,NOARP> mtu 72
unspec 00-00-00-00-00-00-00-00-00-00-00-00-00-00-00-00 txqueuelen 1000 (UNSPEC)
RX packets 8022741 bytes 64181928 (64.1 MB)
RX errors 0 dropped 18 overruns 0 frame 0
TX packets 197508 bytes 1580064 (1.5 MB)
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0
$ ifconfig vxcan888
vxcan888: flags=193<UP,RUNNING,NOARP> mtu 72
unspec 00-00-00-00-00-00-00-00-00-00-00-00-00-00-00-00 txqueuelen 1000 (UNSPEC)
RX packets 197508 bytes 1580064 (1.5 MB)
RX errors 0 dropped 18 overruns 0 frame 0
TX packets 8024709 bytes 64197672 (64.1 MB)
TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0
C 模拟1000个传感器
详细见github链接中的 src/fake_mtlt305d.c, main函数节选
int main(int argc, char **argv)
{
if (argc != 2) {
printf("Usage: %s <numbers>n", argv[0]);
return -1;
}
int n = atoi(argv[1]);
int s[65536];
for(int i = 0; i < n; i++) {
char c[10];
sprintf(c, "vxcan%d", i);
s[i] = socketcan_init(c);
}
struct mtlt305d_t mtlt305d;
struct timespec t0, t1;
int cnt = 0;
while(1) {
clock_gettime(CLOCK_MONOTONIC, &t0);
for(int i = 0; i < n; i++) {
mtlt305d.accx = 1 + cnt % 10;
mtlt305d.accy = 2 + cnt % 10;
mtlt305d.accz = 3 + cnt % 10;
mtlt305d.gyrox = 4 + cnt % 10;
mtlt305d.gyroy = 5 + cnt % 10;
mtlt305d.gyroz = 6 + cnt % 10;
mtlt305d.pitch = 7 + cnt % 10;
mtlt305d.roll = 8 + cnt % 10;
mtlt305d_send(s[i], &mtlt305d);
}
cnt++;
clock_gettime(CLOCK_MONOTONIC, &t1);
double dt = t1.tv_sec - t0.tv_sec + (t1.tv_nsec - t0.tv_nsec) / 1e9;
if(dt < 0.01) {
usleep(10000 - dt*1e6);
}
}
for(int i = 0; i < n; i++) {
close(s[i]);
}
return 0;
}
编译运行
gcc fake_mtlt305d.c mtlt305d.c -o a.out
./a.out 1000
正常写法, 未优化, 单线程, 模拟1000个MTLT305D传感器(10ms出3帧数据, 1000路1s会出300K帧数据, 其实流量不大, 只不过帧数较多), 约占用 50% CPU(i7-8086K, Win11, WSL2, Ubuntu22, kernel 5.15.57.1)
1000路CAN, 假设是500Kbit/s, 那每路CAN的负载率约 9%(约相当于12路CANFD, 5Mbit/s, 100%负载跑满)
$ canbusload can0@500000
can0@500000 297 47520 19008 9%
can0@500000 297 47520 19008 9%
这么看来, 再大胆一点, 开多线程/进程, 在这个6核12线程的CPU模拟10000个这种传感器还是问题不大的.
C 解析1000个传感器
解析就肯定不能像之前那样同步阻塞read了, 开1000个线程可能会被人锤, 这里改用epoll方式.
用C代码(epoll方式)解析1000路传感器数据(打印其中一路), 详细见 src/parser_mtlt305d.c(如有错误, 请留言指正), 循环部分节选
while(running) {
num_events = epoll_wait(fd_epoll, events_pending, n, -1);
if(num_events == -1) {
if (errno != EINTR)
running = 0;
continue;
}
for (int i = 0; i < num_events; i++) {
int *fs = (int *)events_pending[i].data.ptr;
int idx;
/* these settings may be modified by recvmsg() */
iov.iov_len = sizeof(frame);
msg.msg_namelen = sizeof(addr);
msg.msg_controllen = sizeof(ctrlmsg);
msg.msg_flags = 0;
int nbytes = recvmsg(*fs, &msg, 0);
// 第一次会比较慢??
idx = idx2dindex(addr.can_ifindex, *fs);
if (nbytes == -1) {
if (errno != EINTR)
running = 0;
continue;
}
// print can999 parser results
static char name[16] = "can999";
if(strncmp(devname[idx], name, sizeof(name)) == 0) {
int ret = mtlt305d_parser(&frame, &mtlt305d);
if (ret == MTLT305D_ACEINNA_ANGLE_RATE_FRAME_ID) {
struct timespec t;
clock_gettime(CLOCK_REALTIME, &t);
printf("%ld.%09ldt%ft%ft%ft%ft%ft%ft%ft%fn",
t.tv_sec, t.tv_nsec,
mtlt305d.accx, mtlt305d.accy, mtlt305d.accz,
mtlt305d.gyrox, mtlt305d.gyroy, mtlt305d.gyroz,
mtlt305d.pitch, mtlt305d.roll);
}
}
}
编译运行
gcc parser_mtlt305d.c mtlt305d.c -o b.out
./b.out 1000
解析器CPU占用约72%(注释掉打印可降低约10%占用), 随着解析器的运行, 模拟器负载率略有上升(单个CAN通道每多一个订阅, 发送端都要多投递/拷贝)
cpp中的asio封装有epoll, 也能达到类似的解析效果
Github
rust_note/playground/play_1000_cans/src at main · weifengdq/rust_note (github.com)
原文始发于微信公众号（weifengdq）：1000 路 CAN 收发测试
