当前位置：首页 > news >正文

总结了一下中继引擎（can中继器，TCP总机器）开发实际经验

news 来源：原创 2024/5/18 16:46:05

多路数据进行中继的研究

1.数据中继的概念

数据中继是一种数据传输技术，用于在两个通信设备之间提供数字信号的传输。它利用数字信道传输数据信号，可以提供永久性和半永久性连接的数字数据传输信道。

数据中继的主要作用是提高通信质量和可靠性，同时实现多路复用，即在同一个物理链路上传输多个信号。

在数字通信网络中，数据中继可以用于计算机之间的通信，传送数字化传真、数字话音、数字图像信号或其它数字化信号等。

简单来说：中继的核心就是数据传输，比如传输简单的基础数据、话音、传真、图像信息等；

最简单就是1到2和2到1的数据交互，如下模型，就是左右之间数据的交互。

2.中继扩展

在简单的1<>2工作模型，扩展开来，比如3<>4, 5<>6...

等成千上万，上百万个job处理时。就是中继引擎

如上，左边的方式，适合于生命周期短的方案（倾向于通道互斥）；

右边的方式时适合于生命周期长的方案（倾向于通道的信息共享）；

3.中继方案1

实现步骤1

最简单就是1到2和2到1的数据交互的模型实现

拆分读写：

可以用epoll/select/poll 模型都可以；

举例如下：


#include<stdio.h>
#include<stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include<errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/epoll.h>
#define     TTY1       "/dev/tty11"
#define     TTY2       "/dev/tty12"
#define     BUFSIZE     1024
enum {//几种状态STATE_R,STATE_W,STATE_AUTO,STATE_EX,STATE_T
};
struct fsm_st 
{int state;//记录状态int sfd;//源文件int dfd;//目的文件char buf[BUFSIZE];//中间缓冲区int len;//读到的长度int pos;//写的过程如果一次没有写完，记录上次写的位置char* err;//错误信息
};
static void fsm_driver(struct fsm_st* fsm) 
{int ret;switch (fsm->state) {case STATE_R:fsm->len = read(fsm->sfd, fsm->buf, BUFSIZE);if (fsm->len == 0)fsm->state = STATE_T;else if (fsm->len < 0) {if (errno == EAGAIN)fsm->state = STATE_R;else {fsm->err = "read()";fsm->state = STATE_EX;}}else {fsm->pos = 0;fsm->state = STATE_W;}break;case STATE_W:ret = write(fsm->dfd, fsm->buf + fsm->pos, BUFSIZE);if (ret < 0) {if (errno == EAGAIN)fsm->state = STATE_W;else {fsm->err = "write()";fsm->state = STATE_EX;}}else {fsm->pos += ret;fsm->len -= ret;if (fsm->len == 0)fsm->state = STATE_R;//写完了再去读elsefsm->state = STATE_W;//没写完继续写}break;case STATE_EX:perror(fsm->err);fsm->state = STATE_T;break;case STATE_T:/*  do smoething*/break;default:abort();break;}
}
static int max(int a, int b) 
{return a > b ? a : b;
}
static void relay(int fd1, int fd2) 
{struct fsm_st fsm12, fsm21;int epfd;struct epoll_event ev;int fd1_save = fcntl(fd1, F_GETFL);fcntl(fd1, F_SETFL, fd1_save | O_NONBLOCK);		//非阻塞	打开int fd2_save = fcntl(fd2, F_GETFL);fcntl(fd2, F_SETFL, fd2_save | O_NONBLOCK);		//非阻塞	打开//初始状态fsm12.state = STATE_R;fsm12.sfd = fd1;fsm12.dfd = fd2;fsm21.state = STATE_R;fsm21.sfd = fd2;fsm21.dfd = fd1;/*******创建epoll实例：告诉内核这个监听的数目是10*/epfd = epoll_create(10);if (epfd < 0) {perror("epoll_create()");exit(1);}/*控制、设置:注册要监听的事件类型*/ev.events = 0;//暂时不确定监视何种行为 - 位图清0 ev.data.fd = fd1;epoll_ctl(epfd, EPOLL_CTL_ADD, fd1, &ev);//增删改操作:加入fd1-确保系统监视fd1ev.events = 0;//暂时不确定监视何种行为 - 位图清0 ev.data.fd = fd2;epoll_ctl(epfd, EPOLL_CTL_ADD, fd2, &ev);//增删改操作  在循环外添加避免循环中来回操作while (fsm12.state != STATE_T || fsm21.state != STATE_T) {//1.为fd1布置监视任务ev.data.fd = fd1;ev.events = 0;//位图清0if (fsm12.state == STATE_R)   //1->2:说明如果fd1可读ev.events |= EPOLLIN;//读事件if (fsm21.state == STATE_W)   //2->1:说明如果fd1可写ev.events |= EPOLLOUT;//写事件epoll_ctl(epfd, EPOLL_CTL_MOD, fd1, &ev);//增删改操作ev.data.fd = fd2;ev.events = 0;//位图清0if (fsm12.state == STATE_W)//1->2:说明如果fd2可写ev.events |= EPOLLOUT;if (fsm21.state == STATE_R)//2->1:说明如果fd1可读ev.events |= EPOLLIN;//epoll_ctl(epfd, EPOLL_CTL_MOD, fd2, &ev);//增删改操作//2.监视--等待事件发生if (fsm12.state < STATE_AUTO || fsm21.state < STATE_AUTO) {					//epfd组，事件结构体，事件数量-元素个数ev，-1-死等while (epoll_wait(epfd, &ev, 1, -1) < 0) {			 //if (errno == EINTR) //假错{continue;}perror("epoll_wait()");exit(1);}}//3.查看监视结果//监视结果是fd1 且fd1可读if (ev.data.fd == fd1 && ev.events & EPOLLIN || ev.data.fd == fd2 \&& ev.events & EPOLLOUT || fsm12.state > STATE_AUTO)fsm_driver(&fsm12);//如果1可读2可写或者处于EX,T态if (ev.data.fd == fd1 && ev.events & EPOLLOUT || ev.data.fd == fd2 \&& ev.events & EPOLLIN || fsm21.state > STATE_AUTO)//如果2可读或者1可写fsm_driver(&fsm21);}//复原退出fcntl(fd1, F_SETFL, fd1_save);fcntl(fd2, F_SETFL, fd2_save);close(epfd);
}
int main(int argc, char** argv) 
{int fd1, fd2;fd1 = open(TTY1, O_RDWR);//先以阻塞打开（故意先阻塞形式）if (fd1 < 0) {perror("open()");exit(1);}write(fd1, "TTY1\n", 5);fd2 = open(TTY2, O_RDWR | O_NONBLOCK);//非阻塞if (fd2 < 0) {perror("open()");exit(1);}write(fd2, "TTY2\n", 5);relay(fd1, fd2);		//核心代码close(fd2);close(fd1);exit(0);
}

实现步骤2 将该步骤，扩展

每一个job加入数组进行管理，然后遍历数组，实现管理每一个job任务(下方例子仅作参考，有待对每个job加入epoll)

#include<stdio.h>
#include<stdlib.h>
#include<errno.h>
#include<pthread.h>
#include <unistd.h>
#include<string.h>
#include<fcntl.h>
#include"relayer.h"
#define   BUFSIZE  1024
static struct rel_job_st* rel_job[REL_JOBMAX];
static pthread_mutex_t mut_rel_job = PTHREAD_MUTEX_INITIALIZER;
static pthread_once_t init_once = PTHREAD_ONCE_INIT;
enum {//状态机的几种状态STATE_R,STATE_W,STATE_EX,STATE_T
};
enum {//job的状态STATE_RUNNING = 1,STATE_CANCELED,STATE_OVER
};
//状态机
struct rel_fsm_st {int state;//记录状态机的状态int sfd;//源文件int dfd;//目的文件char buf[BUFSIZE];//中间缓冲区int len;//读到的长度int pos;//写的过程如果一次没有写完，记录上次写的位置char* err;//错误信息int64_t count; //输出字符数量
};
//每一对终端结构体
struct rel_job_st{int fd1;//两个终端int fd2;//该对终端状态STATE_RUNNING，STATE_CANCELED, STATE_OVERint job_state;//两个终端的状态机结构体struct rel_fsm_st fsm12, fsm21;//用来退出复原状态int fd1_save, fd2_save;
};
//状态转移函数
static void fsm_driver(struct rel_fsm_st* fsm) {int ret;switch (fsm->state) {case STATE_R:fsm->len = read(fsm->sfd, fsm->buf, BUFSIZE);if (fsm->len == 0)fsm->state = STATE_T;else if (fsm->len < 0) {if (errno == EAGAIN)fsm->state = STATE_R;else {fsm->err = "read()";fsm->state = STATE_EX;}}else {fsm->pos = 0;fsm->state = STATE_W;}break;case STATE_W:ret = write(fsm->dfd, fsm->buf + fsm->pos, fsm->len);if (ret < 0) {if (errno == EAGAIN)fsm->state = STATE_W;else {fsm->err = "write()";fsm->state = STATE_EX;}}else {fsm->pos += ret;fsm->len -= ret;if (fsm->len == 0)fsm->state = STATE_R;//写完了再去读elsefsm->state = STATE_W;//没写完继续写}break;case STATE_EX:perror(fsm->err);fsm->state = STATE_T;break;case STATE_T:/*  do smoething*/break;default:abort();break;}
}
static void *thr_relayer(void *p) {int i;while (1) {pthread_mutex_lock(&mut_rel_job);//死等for (i = 0; i < REL_JOBMAX; i++) {if (rel_job[i] != NULL) //不断的找到一个任务然后推送执行{if (rel_job[i]->job_state == STATE_RUNNING)//运行态{fsm_driver(&rel_job[i]->fsm12);//先推再判断fsm_driver(&rel_job[i]->fsm21);if (rel_job[i]->fsm12.state == STATE_T && rel_job[i]->fsm21.state == STATE_T)rel_job[i]->job_state = STATE_OVER;}}}pthread_mutex_unlock(&mut_rel_job);}}
static void module_load(void) 
{int err;pthread_t tid_relayer;err = pthread_create(&tid_relayer, NULL, thr_relayer, NULL);if (err) {fprintf(stderr, "pthread_create():%s\n", strerror(err));exit(1);}
}
static int get_free_pos_unlocked() {int i;for (i = 0; i < REL_JOBMAX; i++) {if (rel_job[i] == NULL)return i;}return -1;
}
int rel_addjob(int fd1, int fd2) {struct rel_job_st *me;int pos;pthread_once(&init_once, module_load);//单次调用:pthread_onceme = malloc(sizeof(*me));if (me == NULL)   //空间问题return -ENOMEM;me->fd1 = fd1;me->fd2 = fd2;me->job_state = STATE_RUNNING;//该对终端设置正在运行me->fd1_save = fcntl(me->fd1, F_GETFL);fcntl(me->fd1, F_SETFL, me->fd1_save | O_NONBLOCK);  //非阻塞	打开me->fd2_save = fcntl(me->fd2, F_GETFL);fcntl(me->fd2, F_SETFL, me->fd2_save | O_NONBLOCK);//非阻塞	打开me->fsm12.sfd = me->fd1;me->fsm12.dfd = me->fd2;me->fsm12.state = STATE_R;me->fsm21.sfd = me->fd2;me->fsm21.dfd = me->fd1;me->fsm21.state = STATE_R;pthread_mutex_lock(&mut_rel_job);pos = get_free_pos_unlocked();//临界状态-需要加入互斥锁if (pos < 0) {pthread_mutex_unlock(&mut_rel_job);fcntl(me->fd1, F_SETFL, me->fd1_save);//恢复现场fcntl(me->fd2, F_SETFL, me->fd2_save);free(me);//释放空间return -ENOSPC;}rel_job[pos] = me;pthread_mutex_unlock(&mut_rel_job);return pos;
}
int rel_canceljob(int id);
/*return == 0			成功，指定任务成功取消== -EINVAL	失败，参数非法== -EBUSY    失败，任务早已被取消
*/
int rel_waitjob(int id, struct rel_stat_st*);
int rel_statjob(int id, struct rel_stat_st*);

4.中继方案2

方案2不采用方案1的经验，是因为方案2，在多路io复用的时候，容易形成“群惊”效应。

还有，方案1，是通道排斥，2是信息分享；

因此，采用如下方式：每一个收发都配置一个接收队列，发送直接发给对方的队列装载

实现步骤1

队列的实现（队列的实现--数量少采用静态数组/数量多动态数组（进程内），进程间的处理反射光hi，采用共享内存+数组方式）

队列实现-例子：数量少采用静态数组（进程内）

/****************************************************
* 函数名：udp_msg_enqueue
* 功能描述：UDP消息入队
* 输入参数：msg_type-消息类型，msg_in_data-消息数据
* 输出参数：
* 返回值：
* 备注：Fhsj YJ
****************************************************/
int udp2_msg_enqueue(const UDP_MSG_STRU *msg_in_data) {if (NULL == msg_in_data) {printf("func:%s : 指针为NULL", __func__);return 0;}// 数据入队并返回结果if (((udp2_msg_queue.rear + 1) % MAX_UDP_MSG_NUM) == udp2_msg_queue.front) {return 0;} else {memset(&udp2_msg_queue.date[udp2_msg_queue.rear], 0, sizeof(UDP_MSG_STRU));memcpy(&udp2_msg_queue.date[udp2_msg_queue.rear], msg_in_data, sizeof(UDP_MSG_STRU));udp2_msg_queue.rear = (udp2_msg_queue.rear + 1) % MAX_UDP_MSG_NUM;return 1;}
}/****************************************************
* 函数名：dsrc_msg_dequeue
* 功能描述：UDP消息出队
* 输入参数：msg_type-消息类型
* 输出参数：
* 返回值：msg_in_data-消息数据
* 备注：Fhsj YJ
****************************************************/
UDP_MSG_STRU *udp2_msg_dequeue(void) {u16 u16obj_num = 0;// 数据出队if (udp2_msg_queue.rear == udp2_msg_queue.front) {return NULL;} else {u16obj_num = udp2_msg_queue.front;udp2_msg_queue.front = (udp2_msg_queue.front + 1) % MAX_UDP_MSG_NUM;return &udp2_msg_queue.date[u16obj_num];}
}/****************************************************
* 函数名：get_udp_msg_num
* 功能描述：UDP消息数量获取
* 输入参数：msg_type-消息类型
* 输出参数：
* 返回值：消息数量
* 备注：Fhsj YJ
****************************************************/
u8 get_udp2_msg_num(void) {u8 msg_num;if (udp2_msg_queue.rear >= udp2_msg_queue.front) {msg_num = udp2_msg_queue.rear - udp2_msg_queue.front;} else {msg_num = udp2_msg_queue.rear + MAX_UDP_MSG_NUM - udp2_msg_queue.front;}return msg_num;
}