#include "sys_config.h" #include "tx_platform.h" #include "list.h" #include "typesdef.h" #include "osal/string.h" #include "stream_frame.h" #include "utlist.h" #include "osal/work.h" #include "osal/irq.h" //#define STREAM_FRAME_DEBUG_BIND #define STREAM_FRAME_DEBUG_GC static int unbind_stream(stream *s); typedef void *(*stream_malloc)(int size); typedef void (*stream_free)(void *ptr); stream *g_s = NULL; stream *d_s = NULL; static void *extern_malloc(int size) { return os_malloc(size); } static void extern_free(void *ptr) { if (ptr) { os_free(ptr); } } stream_malloc self_malloc = extern_malloc; stream_free self_free = extern_free; static int default_free_data_func(struct data_structure *d) { int res = 0; uint32_t flags; int ref; flags = disable_irq(); ref = --d->ref; enable_irq(flags); if(ref < 0) { os_printf("%s:%d!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!d:%X\n",__FUNCTION__,__LINE__,d); os_printf("d name:%s\n",d->s->name); } //释放空间 if(ref == 0) { //先调用注册的删除函数,后通知流 if(d->ops && d->ops->free) { d->ops->free(d); } res = d->s->func(d->s,d,STREAM_DATA_FREE); //返回0,代表不需要删除data,如果返回>0,代表data实体已经无效 if(res > 0) { d->data = NULL; } flags = disable_irq(); DL_DELETE(d->s->src_d_list_u,d); d->prev = NULL; d->next = NULL; DL_APPEND(d->s->src_d_list_f, d); enable_irq(flags); d->s->func(d->s,d,STREAM_DATA_FREE_END); } return 0; } static int default_stream_priv_func(stream *s,void *priv,int opcode) { return 0; } static void *creat_stream(const char *name) { uint32_t flags; stream *s = (stream*)self_malloc(sizeof(stream)); if(!s) { goto creat_stream_end; } memset(s,0,sizeof(stream)); s->name = name; s->ref = 0; //代表有人创建了,所以一定是有人在使用 s->used = STREAM_AVAILABLE; s->d_list_f = NULL; s->d_list_u = NULL; s->ref++; s->func = default_stream_priv_func; flags = disable_irq(); //往链表头插入,保护 DL_PREPEND(g_s, s); enable_irq(flags); creat_stream_end: return s; } //查找是否存在对应的流名称 static void *find_stream(const char *name) { stream *s = NULL; stream *ret_s = NULL; uint32_t flags; //保护轮询,并且这个阶段不能进行gc flags = disable_irq(); stream *each_s = g_s; DL_FOREACH(each_s,s) { if(!strcmp(s->name,name)) { //这里应该是要进行回收的,这里就直接返回吧,重新创建一个同样名字的流 //旧的流让它回收,防止有异步的可能 if(s->ref == 0) { s->name = NULL; goto find_stream_end; } s->open_ref++; s->ref++; ret_s = s; goto find_stream_end; } } find_stream_end: //轮询完毕,可以进行gc操作(gc的时候要全局保护,先对链表进行移动,最后单独进行回收) enable_irq(flags); return ret_s; } uint8_t find_stream_enable(const char *name) { stream *s = NULL; stream *ret_s = NULL; uint32_t flags; //保护轮询,并且这个阶段不能进行gc flags = disable_irq(); stream *each_s = g_s; DL_FOREACH(each_s,s) { if(!strcmp(s->name,name)) { ret_s = s; break; } } //轮询完毕,可以进行gc操作(gc的时候要全局保护,先对链表进行移动,最后单独进行回收) enable_irq(flags); s = ret_s; if(s && s->enable) return 1; return 0; } //开放的接口 //对流的data空间进行分配,暂定buf是连续的,后续可以增加buf不连续的接口(避免申请一大片的空间) void stream_data_mem_dis(stream *s,uint8_t *buf_head,int uint_len,int count) { struct data_structure *elt; DL_FOREACH(s->src_d_list_f,elt) { if(count == 0) { break; } elt->data = buf_head; buf_head += uint_len; count--; } } //对data进行分配空间,用回调函数操作,count是分配的最大值 void stream_data_dis_mem(stream *s,int count) { struct data_structure *elt; int data_count = 0; DL_FOREACH(s->src_d_list_f,elt) { if(count == 0) { break; } elt->priv = (void*)data_count++; s->func(s,elt,STREAM_DATA_DIS); count--; } } //对data进行分配空间(自定义),用回调函数操作 void stream_data_dis_mem_custom(stream *s) { struct data_structure *elt; int data_count = 0; DL_FOREACH(s->src_d_list_f,elt) { elt->priv = (void*)data_count++; s->func(s,elt,STREAM_DATA_DIS); } } //强行引用加1,特殊作用,适合一次性占用多次,并且不需要经过find_stream的流程,前提流是可用的 int open_stream_again(stream *s) { uint32_t flags; int res = 0; flags = disable_irq(); if(s->used == STREAM_ISUED) { s->open_ref++; s->ref++; } else { res = 1; } enable_irq(flags); return 0; } //从流的名称获取一个已经在运行的流 //配合close_stream使用 void *get_stream_available(const char *name) { stream *s = NULL; uint32_t flags; //检查是否已经创建过了 s = find_stream(name); if(s) { //判断一下状态,如果是在使用,则代表被创建过,直接返回 if(s->used == STREAM_ISUED) { goto get_stream_available_end; } else { flags = disable_irq(); s->open_ref--; s->ref--; enable_irq(flags); s = NULL; } } get_stream_available_end: return s; } //打开一个可用的流,如果没有,则创建一个,如果已经存在,则使用原来的,则参数除了name,其他无效 void *open_stream_available(const char *name,int data_count,int recv_count,stream_priv_func func,void *priv) { uint32_t flags; stream *s = NULL; //检查是否已经创建过了 s = find_stream(name); if(s) { //判断一下状态,如果是在使用,则代表被创建过,直接返回 if(s->used == STREAM_ISUED) { goto open_stream_available_end; } else { flags = disable_irq(); s->open_ref--; s->ref--; enable_irq(flags); } } s = (stream *)open_stream(name,data_count,recv_count,func,priv); open_stream_available_end: return s; } //打开对应的流,如果需要生产data,则data_count>0,如果只是接收,recv_count>0,根据需求来填写 void *open_stream(const char *name,int data_count,int recv_count,stream_priv_func func,void *priv) { stream *s = NULL; void *src_d_list_head = NULL; void *d_list_head = NULL; uint32_t flags; int res = 0; if(func) { res = func(NULL,priv,STREAM_OPEN_ENTER); if(res) { goto open_stream_end; } } //先申请对应的空间 if(data_count > 0) { src_d_list_head = (void*)self_malloc(data_count*sizeof(struct data_structure)); if(src_d_list_head) { memset(src_d_list_head,0,data_count*sizeof(struct data_structure)); } } if(recv_count > 0) { d_list_head = (void*)self_malloc(recv_count*sizeof(struct data_list)); if(d_list_head) { memset(d_list_head,0,recv_count*sizeof(struct data_list)); } } if((data_count>0 && !src_d_list_head) || (recv_count > 0 && !d_list_head)) { if(src_d_list_head) { self_free(src_d_list_head); } if(d_list_head) { self_free(d_list_head); } src_d_list_head = NULL; d_list_head = NULL; goto open_stream_end; } if(func) { res = func(NULL,(void*)name,STREAM_OPEN_CHECK_NAME); if(res) { goto open_stream_end; } } //检查是否已经创建过了 s = find_stream(name); if(s) { flags = disable_irq(); //stream不是可用或者不是重复可用,则返回空,代表名称有了,可能没有及时回收,需要等待及时回收才可以创建同一个名字的流 if(s->used != STREAM_AVAILABLE_AGAIN && s->used != STREAM_AVAILABLE) { s->open_ref--; s->ref--; enable_irq(flags); s = NULL; goto open_stream_end; } //没有被使用,则初始化对应数据 s->used = STREAM_ISUED; enable_irq(flags); } //没有找到对应的流,则创建 else { //这是第一次创建,不需要锁 s = creat_stream(name); if(s) { s->open_ref++; s->used = STREAM_ISUED; } } //创建对应data_list_f数量,相当于初始化 struct data_structure *tmp_data_s = (struct data_structure *)src_d_list_head; for(int i=0;isrc_d_list_f, tmp_data_s); tmp_data_s->free = default_free_data_func; tmp_data_s->s = s; tmp_data_s++; } struct data_list *tmp_data_list = (struct data_list *)d_list_head; for(int i=0;id_list_f, tmp_data_list); tmp_data_list++; } open_stream_end: if(s) { s->src_d_list_count = data_count; s->src_d_list_head = src_d_list_head; s->d_list_head = d_list_head; //这里可以接入一个调试流,用于接收各种可能需要的数据,如果不需要,也只是浪费一点内存空间 streamSrc_bind_streamDest(s,R_DEBUG_STREAM); if(func) { s->func = func; func(s,priv,STREAM_OPEN_EXIT); } else { s->func = default_stream_priv_func; } } else { //如果没有流,看看是否申请了空间 if(src_d_list_head) { self_free(src_d_list_head); } if(d_list_head) { self_free(d_list_head); } if(func) { func(s,priv,STREAM_OPEN_FAIL); } } return s; } int close_stream(stream *s) { if(!s) { return 1; } uint32_t flags; uint32_t ref; flags = disable_irq(); ref = --s->open_ref; //如果流还有被打开过,则需要等待另一个占用被释放 if(ref) { s->ref--; //理论这里不可能等于0 enable_irq(flags); return 2; } struct data_list *elt,*tmp; struct data_list *del_u; s->enable = 0; //关闭使能 s->used = STREAM_UNAVAILABLE; //代表不可使用 del_u = s->d_list_u; s->d_list_u = NULL; enable_irq(flags); s->func(s,NULL,STREAM_CLOSE_ENTER); //释放一下自己接收到的数据链表内容,然后它就没有接收的数据了 DL_FOREACH_SAFE(del_u,elt,tmp) { //从已有链表释放 DL_DELETE(del_u,elt); //进行elt->data删除 free_data(elt->data); //放回到空闲链表 elt->prev = NULL; elt->next = NULL; flags = disable_irq(); DL_APPEND(s->d_list_f,elt); enable_irq(flags); } s->func(s,NULL,STREAM_CLOSE_EXIT); s->ref--; s->used = STREAM_WAIT_GC_MARKER; //等待gc标记处理 //统计src_d_list_f是否与count一致(如果不一致,可能存在悬空态,但未实际发送到流,正常不应该出现这个情况),实际是判断src_d_list_u是否还有被使用,如果有,代表还不能释放 //剩下交给gc去处理而不是在这里处理 return 0; } //内部使用解锁绑定 static int unbind_stream(stream *s) { if(!s) { return 1; } uint32_t flags; #ifdef STREAM_FRAME_DEBUG_BIND os_printf("%s name:%s\tref:%d\n",__FUNCTION__,s->name,s->ref); #endif flags = disable_irq(); s->ref--; //理论这里不可能等于0 enable_irq(flags); return 0; } //s为数据产生的源头,然后绑定对应的name int streamSrc_bind_streamDest(stream *s,const char *name) { int res = -1; stream *dest = NULL; struct stream_list *s_list; uint32_t flags; if(!s) { return res; } dest = find_stream(name); if(dest) { //保证是可用的 if(dest->used >= 0) { //检查一下是否已经绑定过,绑定过就不要重复绑定了 flags = disable_irq(); DL_FOREACH(s->s_list,s_list) { if(s_list->s == dest) { res = 0; dest->ref--; dest->open_ref--; break; } } enable_irq(flags); if(!res) { os_printf("alread %s:%d\n",__FUNCTION__,__LINE__); goto streamSrc_bind_streamDest_finish; } //没有绑定过 s_list = (struct stream_list *)self_malloc(sizeof(struct stream_list)); if(s_list) { memset(s_list,0,sizeof(struct stream_list)); s_list->s = dest; flags = disable_irq(); DL_PREPEND(s->s_list, s_list); //往列表前面添加 enable_irq(flags); res = 0; } //空间不足 else { flags = disable_irq(); dest->ref--; enable_irq(flags); } } else { dest->ref--; } flags = disable_irq(); dest->open_ref--; enable_irq(flags); } //如果name不存在,则创建一个空的stream s else { dest = creat_stream(name); if(dest) { s_list = (struct stream_list *)self_malloc(sizeof(struct stream_list)); if(s_list) { memset(s_list,0,sizeof(struct stream_list)); s_list->s = dest; flags = disable_irq(); DL_PREPEND(s->s_list, s_list); //往列表前面添加 enable_irq(flags); res = 0; } } } streamSrc_bind_streamDest_finish: return res; } #if 0 //解绑定 int streamSrc_unbind_streamDest(stream *s,const char *name) { int res = -1; struct stream_list *elt,*tmp; stream *dest = NULL; dest = find_stream(name); if(dest) { DL_FOREACH_SAFE(s->s_list,elt,tmp) { if(dest == elt->s) { DL_DELETE(s->s_list,elt); dest->ref--; self_free(elt); res = 0; break; } } uint32_t flags; flags = disable_irq(); //find_stream的时候ref增加了,所以要-- dest->ref--; dest->open_ref--; enable_irq(flags); } else { } return res; } #endif //hold住data(保证data不会被删除),不让data被删除,可以调用free_data来解锁 int hold_data(struct data_structure *data) { uint32_t flags; flags = disable_irq(); data->ref++; enable_irq(flags); return 0; } //向stream中发送data数据,会轮询s_list链表发送 int send_data_to_stream(struct data_structure *data) { struct stream_list *elt; struct data_list *d_list,*tmp; int count = 0; //计算发送到多少个数据流 uint32_t flags; data->ref++; flags = disable_irq(); //放到已经使用的链表中 DL_APPEND(data->s->src_d_list_u, data); enable_irq(flags); data->s->func(data->s,data,STREAM_SEND_DATA_START); flags = disable_irq(); struct stream_list *each_list = data->s->s_list; //防止s_list被修改的可能,因为绑定是放在链表头,错过就是下一次再发送,不至于线程切换导致链表有可能出问题 enable_irq(flags); DL_FOREACH(each_list,elt) { elt->s->send_ref++; if(elt->s->enable && elt->s->used == STREAM_ISUED) { //先判断一下该数据是否要发送到流,如果不需要,则不需要进行往下操作 if(elt->s->func(elt->s,data,STREAM_FILTER_DATA)) { continue; } d_list = NULL; //获取elt->s中可以使用的数据链表,如果没有则不发送过去,代表对方空间不够 flags = disable_irq(); DL_FOREACH_SAFE(elt->s->d_list_f,d_list,tmp) { DL_DELETE(elt->s->d_list_f,d_list); break; } enable_irq(flags); if(d_list) { data->s->func(data->s,data,STREAM_SEND_TO_DEST); count++; data->ref++; d_list->data = data; //链表的next与prev清空 d_list->next = NULL; d_list->prev = NULL; //发送到可用的链表 flags = disable_irq(); DL_APPEND(elt->s->d_list_u, d_list); enable_irq(flags); //调用对应唤醒的接口 elt->s->func(elt->s,data,STREAM_RECV_DATA_FINISH); } } elt->s->send_ref--; } //告诉流,发送数据完成 data->s->func(data->s,data,STREAM_SEND_DATA_FINISH); //进行删除,应该是调用回调函数 if(data->free) { data->free(data); } return count; } //强行进行data删除,可能已经不再需要 int force_del_data(struct data_structure *data) { if(!data) { return 0; } uint32_t flags; data->ref++; //放到已经使用的链表中 flags = disable_irq(); DL_APPEND(data->s->src_d_list_u, data); enable_irq(flags); //进行删除,应该是调用回调函数 if(data->free) { data->free(data); } return 0; } //获取data的空间链表,用于生成数据 struct data_structure *get_src_data_f(stream *s) { struct data_structure *data,*tmp; uint32_t flags; data = NULL; DL_FOREACH_SAFE(s->src_d_list_f,data,tmp) { flags = disable_irq(); DL_DELETE(s->src_d_list_f,data); enable_irq(flags); data->prev = NULL; data->next = NULL; //data->type = 0; break; } return data; } //接收data的链表,用于从其他流接收到数据 struct data_structure *recv_real_data(stream *s) { if(!s) { return NULL; } uint32_t flags; struct data_list *data,*tmp; struct data_structure *real_data = NULL; data = NULL; DL_FOREACH_SAFE(s->d_list_u,data,tmp) { //从used中移除 flags = disable_irq(); DL_DELETE(s->d_list_u,data); enable_irq(flags); data->prev = NULL; data->next = NULL; if(!s->func(s,data->data,STREAM_RECV_FILTER_DATA)) { break; } //代表该数据包需要被过滤 else { flags = disable_irq(); DL_APPEND(s->d_list_f, data); enable_irq(flags); //删除对应的data free_data(data->data); data = NULL; } } if(data) { real_data = data->data; //放回到d_list_f flags = disable_irq(); DL_APPEND(s->d_list_f, data); enable_irq(flags); real_data->s->func(s,real_data,STREAM_CALC_AUDIO_ENERGY); } return real_data; } int free_data(void *data_struct) { struct data_structure *data = (struct data_structure*)data_struct; if(data && data->free) { return data->free(data); } return 0; } //使能stream int enable_stream(stream *s,int enable) { if(!s) { return 1; } s->enable = enable; return 0; } //如果有注册获取get_data函数,则由get_data去获取,否则直接返回data->data的指针 void *get_stream_real_data(struct data_structure* data) { if(data && data->ops && data->ops->get_data) { return data->ops->get_data(data); } else { return data->data; } } uint32_t get_stream_real_data_len(struct data_structure* data) { if(data && data->ops && data->ops->get_data_len) { return data->ops->get_data_len(data); } else { return data->len; } } uint32_t get_stream_data_timestamp(struct data_structure* data) { if(data && data->ops && data->ops->get_data_time) { return data->ops->get_data_time(data); } else { return data->timestamp; } } uint32_t set_stream_real_data_len(struct data_structure* data,uint32_t len) { if(data && data->ops && data->ops->set_data_len) { return data->ops->set_data_len(data,len); } else { data->len = len; return 0; } } uint32_t set_stream_data_time(struct data_structure* data,uint32_t timestamp) { if(data && data->ops && data->ops->set_data_time) { return data->ops->set_data_time(data,timestamp); } else { data->timestamp = timestamp; //默认 return 0; } } uint32_t stream_data_custom_cmd_func(struct data_structure* data,int opcode,void *priv) { if(data && data->ops && data->ops->custom_func) { return data->ops->custom_func(data,opcode,priv); } else { return 0; } } uint32_t register_stream_self_cmd_func(stream *s,stream_self_cmd func) { s->self_cmd_func = func; return 0; } uint32_t stream_self_cmd_func(stream *s,int opcode,uint32_t arg) { if(s && s->self_cmd_func) { return s->self_cmd_func(s,opcode,arg); } //没有对应的函数 return 0; } void broadcast_cmd_to_destStream(stream *s,uint32_t cmd) { struct stream_list *elt; uint32_t flags; flags = disable_irq(); struct stream_list *each_list = s->s_list; //防止s->s_list被修改的可能 enable_irq(flags); DL_FOREACH(each_list,elt) { if(elt->s->enable && elt->s->used == STREAM_ISUED) { elt->s->func(elt->s,(void*)cmd,STREAM_SEND_CMD); } } } int stream_gc_marker_timer(uint32_t t) { stream *s; stream *each_s = g_s; //对链表进行轮询,检查是否要删除,如果需要删除,则进行标记,在另一个workqueue进行回收,与这个是同一个线程(这样不用考虑一些) DL_FOREACH(each_s,s) { if(s->used == STREAM_WAIT_GC_CLEAN) { continue; } //如果被标记或者ref为0,则会自动启动gc处理 if(s->used == STREAM_WAIT_GC_MARKER || s->ref == 0) { if(s->send_ref) { continue; } if(s->d_list_u) { struct data_list *elt,*tmp; uint32_t flags; DL_FOREACH_SAFE(s->d_list_u,elt,tmp) { //从已有链表释放 DL_DELETE(s->d_list_u,elt); //进行elt->data删除 free_data(elt->data); //放回到空闲链表 elt->prev = NULL; elt->next = NULL; flags = disable_irq(); DL_APPEND(s->d_list_f,elt); enable_irq(flags); } } //删除自己bind的stream if(s->s_list) { struct stream_list *elt,*tmp; DL_FOREACH_SAFE(s->s_list,elt,tmp) { DL_DELETE(s->s_list,elt); //close_stream(elt->s); unbind_stream(elt->s); //elt->s->ref--; self_free(elt); } s->s_list = NULL; } //要删除,但是看一下对应的资源是否释放完成 //统计src_d_list_f是否与count一致(如果不一致,可能存在悬空态, //但未实际发送到流,正常不应该出现这个情况),实际是判断src_d_list_u是否还有被使用,如果有,代表还不能释放 if(!s->src_d_list_u) { //轮询src_d_list_f,查看是否有需要释放空间的可能 struct data_structure *data_tmp; DL_FOREACH(s->src_d_list_f,data_tmp) { s->func(s,data_tmp,STREAM_DATA_DESTORY); } //可以进行资源释放 if(s->src_d_list_head) { self_free(s->src_d_list_head); s->src_d_list_head = NULL; } if(s->d_list_head) { self_free(s->d_list_head); s->d_list_head = NULL; } s->src_d_list_f = NULL; s->src_d_list_u = NULL; s->d_list_f = NULL; s->d_list_u = NULL; //状态修改 //如果引用为0,则代表这个流不需要继续存在,可以被删除 if(!s->ref) { int status; uint32_t flag = disable_irq(); status = s->used ; s->used = STREAM_WAIT_GC_CLEAN; enable_irq(flag); char *name = s->name; #ifdef STREAM_FRAME_DEBUG_GC //如果这里将name设置为NULL,则可以在被清除之前也可以创建数据,但是有一个问题就是,尽量不要用类似全局,因为可能会推迟释放资源 os_printf("%s wait gc name:%s\t%X\tfunc:%X\n",__FUNCTION__,s->name,s,s->func); #endif s->name = NULL; //如果等于STREAM_AVAILABLE_AGAIN,代表已经执行过一次 if(status != STREAM_AVAILABLE_AGAIN) { s->func(s,NULL,STREAM_MARK_GC_END); } s->used = STREAM_WAIT_GC_CLEAN; //交给下一个阶段的gc进行链表移位,删除 } else { //先回调函数,就是有可能出现used没有被改,STREAM_MARK_GC_END调用完毕后,被调度,导致used状态没有改变不能创建流 //这里影响创建,但只要稍微调度一下,应该就可以正常,因为不想去开关中断,所以这里暂时先这样 s->func(s,NULL,STREAM_MARK_GC_END); //代表资源相关资源被释放过,后续可以重复利用 s->used = STREAM_AVAILABLE_AGAIN; } } } } return t; } //返回下一次回收的时间,如果没有回收成功,则会10ms后重新进行回收 int stream_gc_clean_timer(uint32_t t) { static uint16_t gc_timer = 100; uint32_t flags; stream *el,*tmp; //可以进行回收,要进行全局保护 flags = disable_irq(); DL_FOREACH_SAFE(g_s,el,tmp) { //只是进行链表移位,最后才进行实际删除 if(el->used == STREAM_WAIT_GC_CLEAN) { DL_DELETE(g_s,el); el->next = NULL; el->prev = NULL; DL_PREPEND(d_s, el); } } enable_irq(flags); //链表进行回收 DL_FOREACH_SAFE(d_s,el,tmp) { //name已经被清除,在marker阶段被设置为NULL,所以name暂时不要用申请的空间,用静态(在flash) DL_DELETE(d_s,el); //进行资源的全部回收 //回调删除命令 if(el->func) { el->func(el,NULL,STREAM_DEL); } if ( el->ref != 0 ) { os_printf("s:%x s->ref:%x\n",el,el->ref); } self_free(el); } gc_timer = t; return gc_timer; } static struct os_work gc_marker; static struct os_work gc_clean; static int32 stream_gc_marker_timer_work(struct os_work *work) { uint32_t t; t = stream_gc_marker_timer(5); os_run_work_delay(&gc_marker, t); return 0; } static int32 stream_gc_clean_timer_work(struct os_work *work) { uint32_t t; t = stream_gc_clean_timer(5); os_run_work_delay(&gc_clean, t); return 0; } void stream_work_queue_start() { OS_WORK_INIT(&gc_marker, stream_gc_marker_timer_work, 0); os_run_work_delay(&gc_marker, 1000); OS_WORK_INIT(&gc_clean, stream_gc_clean_timer_work, 0); os_run_work_delay(&gc_clean, 1000); }