#include "../inc/mq.h" #include //for memcpy #include //for malloc free #include //for struct timeval gettimeofday #include //for signal sigset_t sigfillset pthread_sigmask SIG* #include //for msgget msgsnd msgrcv #include //for pipe write read close #include //for pthread* #include //for errno E* #include "../inc/rbtree.h" #include #define QUE_MSGTYPE 1l static int timer_alive = 0; static pthread_t timer_tid; static pthread_mutex_t timer_mutex = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t timer_cond = PTHREAD_COND_INITIALIZER; static rbtree_t timer_root; typedef struct timer_node_s timer_node_t; struct timer_node_s{ pthread_t tid; uint64_t expire; int status;//0 default, 1 timeout rbtree_node_t rbtree_node; }; int is_thread_alive(pthread_t tid){ int kill_rc = pthread_kill(tid,0);//0 success, ESRCH 线程不存在, EINVAL 信号不合法 int ret = 1;//alive if(kill_rc == ESRCH){ ret = 0; } return ret; } static void* timer_loop(void*arg); static int timer_create_new(){ if(timer_alive == 1){ return 0; } pthread_mutex_lock(&timer_mutex); if(timer_alive == 1){ pthread_mutex_unlock(&timer_mutex); return 0; } rbtree_init(&timer_root); pthread_create(&timer_tid,NULL,timer_loop,NULL); timer_alive = 1; pthread_detach(timer_tid);//分离线程 pthread_mutex_unlock(&timer_mutex); return 0; } static void* timer_loop(void*arg){ sigset_t sigset; sigfillset(&sigset); pthread_sigmask(SIG_SETMASK,&sigset,NULL);//当前线程屏蔽所有信号 uint64_t now; struct timespec now_ts; struct timespec timeout_ts; rbtree_node_t *rbtree_node; timer_node_t *timer_node=NULL,*timer_node_will_expire=NULL; for(;;){ //加锁 pthread_mutex_lock(&timer_mutex); clock_gettime(CLOCK_REALTIME, &now_ts); now = (uint64_t)((uint64_t)now_ts.tv_sec*1000) + (uint64_t)((uint64_t)now_ts.tv_nsec/1000000); timer_node = NULL; timer_node_will_expire = NULL; for(rbtree_node = rbtree_min(&timer_root,timer_root.root); rbtree_node!=NULL; rbtree_node = rbtree_next(&timer_root,rbtree_node) ){ timer_node = rbtree_data(rbtree_node,timer_node_t,rbtree_node); //printf("[tim]loop now=%llu,exp=%llu\n",now,timer_node->expire); if(timer_node->expire <= now){ printf("[tim]kill loop now=%llu,exp=%llu\n",now,timer_node->expire); //超时, 则信号通知 timer_node->status = 1; pthread_kill(timer_node->tid,SIGUSR1); continue; } timer_node_will_expire = timer_node; printf("[tim]loop now=%llu,exp=%llu\n",now,timer_node->expire); break; } //条件等待 if(timer_node_will_expire == NULL){ pthread_cond_wait(&timer_cond,&timer_mutex); }else{ //printf("[tim].will=%llu\n",timer_node_will_expire->expire); timeout_ts.tv_sec = (uint64_t)((uint64_t)timer_node_will_expire->expire/1000); timeout_ts.tv_nsec = (uint64_t)((uint64_t)timer_node_will_expire->expire%1000)*1000000; printf("[tim]con sec=%llu,nsec=%llu\n",(uint64_t)timeout_ts.tv_sec, (uint64_t)timeout_ts.tv_nsec); pthread_cond_timedwait(&timer_cond,&timer_mutex,&timeout_ts);//注意, 时间不能是NULL } //解锁 pthread_mutex_unlock(&timer_mutex); } timer_alive = 0; pthread_exit(NULL); } int add_timer(timer_node_t *node,size_t timeout){//单位ms node->tid = pthread_self(); node->status = 0; struct timespec ts; uint64_t lldata; clock_gettime(CLOCK_REALTIME, &ts); node->expire = (uint64_t)(((uint64_t)ts.tv_sec*1000)+ (uint64_t)timeout + (uint64_t)((uint64_t)ts.tv_nsec/1000000)); node->rbtree_node.key = node->expire; lldata = (uint64_t)(((uint64_t)ts.tv_sec*1000) + (uint64_t)((uint64_t)ts.tv_nsec/1000000)); printf("[rec].time_exp=%llu\n",node->expire); printf("[rec].time_cur=%llu\n",lldata); pthread_mutex_lock(&timer_mutex); rbtree_insert(&timer_root,&node->rbtree_node); pthread_cond_signal(&timer_cond); pthread_mutex_unlock(&timer_mutex); //pthread_cond_signal(&timer_cond); return 0; } int delete_timer(timer_node_t *node){ pthread_mutex_lock(&timer_mutex); rbtree_delete(&timer_root,&node->rbtree_node); pthread_mutex_unlock(&timer_mutex); return 0; } int _signal(int signo, void (*sighandler)(int)) { struct sigaction sa; sigset_t set; //sigemptyset(&set);//当进入sighandler时,不屏蔽所有信号,可被其他信号中断 sigfillset(&set);//当进入sighandler时,屏蔽所有信号 sa.sa_mask = set; sa.sa_handler = sighandler; sa.sa_restorer = NULL; sa.sa_flags = 0; return sigaction(signo, &sa, NULL); } void timer_sighandler(int signo){ return; } typedef struct { long msgtype; unsigned char buffer[0]; } MSGBUF; ssize_t send_que_timedwait(key_t qid,void* pdata,size_t len,int timeout){ int msgqid,retu; timer_node_t node = {0}; unsigned char* buf = NULL; MSGBUF* que_msgbuf; //https://man7.org/linux/man-pages/man2/msgget.2.html msgqid=msgget(qid,0666); if(msgqid == -1){ if(errno == ENOENT){ msgqid=msgget(qid,IPC_CREAT|0666); } if(msgqid == -1) return -1; } buf = malloc(sizeof(MSGBUF) + len); if(buf == NULL) return -1; que_msgbuf = (MSGBUF*)buf; que_msgbuf->msgtype = QUE_MSGTYPE; memcpy(que_msgbuf->buffer,pdata,len); // int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg) // msgflag // 0:当消息队列满时,msgsnd将会阻塞,直到消息能写进消息队列 // IPC_NOWAIT:当消息队列已满的时候,msgsnd函数不等待立即返回 // IPC_NOERROR:若发送的消息大于size字节,则把该消息截断,截断部分将被丢弃,且不通知发送进程。 // 返回值 // 成功:0 // 出错:-1,错误原因存于error中 // EAGAIN:参数msgflg设为IPC_NOWAIT,而消息队列已满 // EIDRM:标识符为msqid的消息队列已被删除 // EACCESS:无权限写入消息队列 // EFAULT:参数msgp指向无效的内存地址 // EINTR:队列已满而处于等待情况下被信号中断 // EINVAL:无效的参数msqid、msgsz或参数消息类型type小于0 if(timeout == -1){ retu=msgsnd(msgqid,buf,len,0); //printf(""); } else if(timeout == 0){ retu=msgsnd(msgqid,buf,len,IPC_NOWAIT); }else{ timer_create_new(); _signal(SIGUSR1,timer_sighandler); add_timer(&node,timeout); retu=msgsnd(msgqid,buf,len,0); delete_timer(&node); if(errno == EINTR && node.status == 1){ errno = ETIMEDOUT; } } free(buf); return retu; } ssize_t read_que_timedwait(key_t qid,void* pdata,size_t len,int timeout){ int msgqid,retu; timer_node_t node = {0}; unsigned char* buf = NULL; MSGBUF* que_msgbuf = NULL; msgqid=msgget(qid,0666);//https://man7.org/linux/man-pages/man2/msgget.2.html if(msgqid == -1){ if(errno == ENOENT){ msgqid=msgget(qid,IPC_CREAT|0666); } if(msgqid == -1) printf("msgget err.\n"); return -1; } buf = malloc(sizeof(MSGBUF) + len); if(buf == NULL){ printf("malloc err.\n"); return -1; } if(timeout == -1){ retu=msgrcv(msgqid,buf,len,0,0); } else if(timeout == 0){ retu=msgrcv(msgqid,buf,len,0,IPC_NOWAIT); }else{ timer_create_new(); _signal(SIGUSR1,timer_sighandler); add_timer(&node,timeout); retu=msgrcv(msgqid,buf,len,0,0); delete_timer(&node); if(errno == EINTR && node.status == 1){ errno = ETIMEDOUT; } } que_msgbuf = (MSGBUF*)buf; if(retu > 0){ memcpy(pdata,que_msgbuf->buffer,retu); } free(buf); return retu; }