1000路CAN收发测试

前言

得益于最近这些年高并发, 高密集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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

这个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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

C 模拟1000个传感器

详细见github链接中的 src/fake_mtlt305d.c, main函数节选

int main(int argc, char **argv)
{
    if (argc != 2) {
        printf("Usage: %s \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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

编译运行

gcc fake_mtlt305d.c mtlt305d.c -o a.out
./a.out 1000
1
2

正常写法, 未优化, 单线程, 模拟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%
1
2
3
4

这么看来, 再大胆一点, 开多线程/进程, 在这个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.%09ld\t%f\t%f\t%f\t%f\t%f\t%f\t%f\t%f\n", \
                        t.tv_sec, t.tv_nsec, \
                        mtlt305d.accx,  mtlt305d.accy,  mtlt305d.accz, \
                        mtlt305d.gyrox, mtlt305d.gyroy, mtlt305d.gyroz, \
                        mtlt305d.pitch, mtlt305d.roll);
                }
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

编译运行

gcc parser_mtlt305d.c mtlt305d.c -o b.out
./b.out 1000
1
2

解析器CPU占用约72%(注释掉打印可降低约10%占用), 随着解析器的运行, 模拟器负载率略有上升(单个CAN通道每多一个订阅, 发送端都要多投递/拷贝)

cpp中的asio封装有epoll, 也能达到类似的解析效果

Github

rust_note/playground/play_1000_cans/src at main · weifengdq/rust_note (github.com)