条件变量的使用：将互斥量的忙等机制改为通知机制
涉及到的函数有以下几个：
int pthread_cond_destroy(pthread_cond_t *cond); /********************** *功能：条件变量的初始化 *参数：cond：条件变量 * attr：条件变量的属性 * ********************/ int pthread_cond_init(pthread_cond_t *restrict cond , const pthread_condattr_t *restrict attr); /***静态初始化***/ pthread_cond_t cond = PTHREAD_COND_INITIALIZER; /*****发送广播给所有处于阻塞等待状态的线程********/ int pthread_cond_broadcast(pthread_cond_t *cond); /******叫醒某一个处于阻塞等待状态的线程*******/ int pthread_cond_signal(pthread_cond_t *cond); int pthread_cond_timedwait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex,const struct timespec *restrict abstime); /********使线程阻塞在一个条件变量上********/ int pthread_cond_wait(pthread_cond_t *restrict cond , pthread_mutex_t *restrict mutex);
eg：
mytbf.c
#include <stdio.h> #include <stdlib.h> #include <sys/types.h> #include <sys/stat.h> #include <fcntl.h> #include <unistd.h> #include <errno.h> #include <pthread.h> #include "mytbf.h" //每个令牌桶内部的信息状态 struct mytbf_st { int cps;//速率 int burst;//上限 int token;//令牌个数 int pos;//令牌桶在这个令牌桶数组中的位置下标 pthread_mutex_t mut;//并发多个线程同时使用成员，保证变量被独占使用，所以每个令牌桶应该有一把锁 pthread_cond_t cond;//通知token累加完成 }; //令牌桶数组 static struct mytbf_st *job[MYTBF_MAX]; //由于令牌桶数组可能被其他并发线程同时访问修改（eg：同时调用init），所以需要添加互斥量 static pthread_mutex_t mut_job = PTHREAD_MUTEX_INITIALIZER; //线程 static pthread_t tid; //只执行一次 static pthread_once_t init_once = PTHREAD_ONCE_INIT; //线程：任务是负责每秒向令牌桶数组中加token static void *thr_alrm(void *p) { int i; while(1) { //令牌桶数组加锁 pthread_mutex_lock(&mut_job); for(i = 0 ; i < MYTBF_MAX; i++) { if(job[i] != NULL) { //锁令牌并自加 pthread_mutex_lock(&job[i]->mut); //加令牌 job[i]->token += job[i]->cps; if(job[i]->token > job[i]->burst) job[i]->token = job[i]->burst; pthread_cond_broadcast(&job[i]->cond); pthread_mutex_unlock(&job[i]->mut); } } //令牌桶解锁 pthread_mutex_unlock(&mut_job); sleep(1); } } //模块的卸载 static void module_unload(void) { int i; //取消线程以及收尸 pthread_cancel(tid); pthread_join(tid,NULL); //将令牌桶释放空间并销毁 pthread_mutex_lock(&mut_job); for(i = 0 ; i < MYTBF_MAX ;i++) { if(job[i] != NULL) { pthread_mutex_destroy(&job[i]->mut); pthread_cond_destroy(&job[i]->cond); free(job[i]); } } pthread_mutex_unlock(&mut_job); pthread_mutex_destroy(&mut_job); return ; } //模块加载 static void module_load(void) { int err; //创建线程 err = pthread_create(&tid,NULL,thr_alrm,NULL); if(err) { fprintf(stderr,"pthread_create():%s ",strerror(err)); exit(1); } //挂载钩子函数：当exit时，调用module_unload atexit(module_unload); } //获取空置的令牌数组位置 static int get_free_pos_unlocked(void) { int i; for(i = 0 ; i < MYTBF_MAX; i++) if(job[i] == NULL) return i; return -1; } //令牌桶初始化 mytbf_t *mytbf_init(int cps,int burst) { struct mytbf_st *me; int pos; //只调用一次module_load pthread_once(&init_once,module_load); //申请空间 me = malloc(sizeof(*me)); if(me == NULL) return NULL; //赋初值 me->cps = cps; me->burst = burst; me->token = 0; pthread_mutex_init(&me->mut,NULL); pthread_cond_init(&me->cond,NULL); //令牌桶加锁查看空位并赋值 pthread_mutex_lock(&mut_job); pos = get_free_pos_unlocked(); if(pos < 0) { free(me); pthread_mutex_unlock(&mut_job); return NULL; } me->pos = pos; job[pos] = me; pthread_mutex_unlock(&mut_job); return me; } static int min(int a,int b) { if(a < b) return a; return b; } //获取令牌个数 int mytbf_fetchtoken(mytbf_t *ptr,int size) { struct mytbf_st *me = ptr; int n; if(size < 0) return -EINVAL; //锁令牌并处理 pthread_mutex_lock(&me->mut); //没有令牌，解决忙等 while(me->token <= 0) { //非忙等 pthread_cond_wait(&me->cond,&me->mut); /*忙等 pthread_mutex_lock(&me->mut); sched_yield(); pthread_mutex_unlock(&me->mut); */ } n = min(me->token,size); me->token -= n; pthread_mutex_unlock(&me->mut); return n; } //返还令牌 int mytbf_returntoken(mytbf_t *ptr,int size) { struct mytbf_st *me = ptr; if(size < 0) return -EINVAL; //加锁处理令牌 pthread_mutex_lock(&me->mut); me->token += size; if(me->token > me->burst) me->token = me->burst; //令牌归还后广播 pthread_cond_broadcast(&me->cond); pthread_mutex_unlock(&me->mut); return 0; } //销毁令牌 int mytbf_destroy(mytbf_t *ptr) { struct mytbf_st *me = ptr; //令牌桶加锁销毁 pthread_mutex_lock(&mut_job); job[me->pos] = NULL; pthread_mutex_unlock(&mut_job); //销毁互斥量和条件变量 pthread_mutex_destroy(&me->mut); pthread_cond_destroy(&me->cond); free(ptr); return 0; }
mytbf.h
+ main.c makefile + mytbf.c mytbf.h X #ifndef MYTBF_H__ #define MYTBF_H__ typedef void mytbf_t; #define MYTBF_MAX 1024 mytbf_t *mytbf_init(int cps,int burst); int mytbf_fetchtoken(mytbf_t *,int size); int mytbf_returntoken(mytbf_t *,int size); int mytbf_destroy(mytbf_t *); #endif
main.c
/****************** *功能：按照流控方式（正常每秒10个字节，有令牌桶机制）从文件中读取数据并写stdout * 使用多线程方式 * 使用互斥量和条件变量 * ***************/ #include <stdio.h> #include <stdlib.h> #include <sys/types.h> #include <sys/stat.h> #include <fcntl.h> #include <unistd.h> #include <errno.h> #include <signal.h> #include <string.h> #include "mytbf.h" //速率：每秒传输10个字节 #define CPS 10 #define BUFSIZE 1024 //令牌上限：100个 #define BURST 100 int main(int argc,char **argv) { int sfd,dfd = 1; char buf[BUFSIZE]; int pos,len,ret; int size; mytbf_t *tbf; //1.判断输入文件路径是否正确合法 if(argc < 2) { fprintf(stderr,"Usage... "); exit(1); } //2.初始化令牌桶：设置速率和令牌上限 tbf = mytbf_init(CPS,BURST); if(tbf == NULL) { fprintf(stderr,"mytbf_init() failed. "); exit(1); } //3.尝试打开文件 do { sfd = open(argv[1],O_RDONLY); if(sfd < 0) { if(errno != EINTR) { perror("open()"); exit(1); } } }while(sfd < 0); //4.读写文件 while(1) { //4.1获取令牌个数 size = mytbf_fetchtoken(tbf,BUFSIZE); if(size < 0) { fprintf(stderr,"mytbf_fetchtoken():%s ",strerror(-size)); exit(1); } //4.2读取相应个数的数据 while((len = read(sfd,buf,size)) < 0) { if(errno == EINTR) continue; perror("read()"); break; } //4.3无数据返回 if(len == 0) break; //4.4有数据判断文件实际读取的长度是否与令牌个数相同 // 如果令牌数量 > 实际读取数量 需要返还令牌 if(size-len > 0) mytbf_returntoken(tbf,size-len); //4.5输出位置定向 pos = 0; //4.6写到输出文件 while(len > 0) { ret = write(dfd,buf+pos,len); if(ret < 0) { if(errno == EINTR) continue; perror("write()"); exit(1); } pos += ret; len -= ret; } } //5.关闭文件，并销毁令牌桶 close(sfd); mytbf_destroy(tbf); exit(0); }
Makefile:
CFLAGS+=-pthread LDFLAGS+=-pthread all:mytbf mytbf:mytbf.o main.o gcc $^ -o $@ $(CFLAGS) $(LDFLAGS) clean: rm -rf *.o mytbf