多路数据进行中继的研究
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;
}
实现步骤2,
每一个fd,对应一个数据处理线程,在线程,在每一个线程收到数据后,进入对应队列
然后再另外一个地方(线程或者进程),出队,处理...
对应代码,已经实现,暂不展示了