条件变量的使用：将互斥量的忙等机制改为通知机制
涉及到的函数有以下几个：
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\n",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...\n");
exit(1);
}
//2.初始化令牌桶：设置速率和令牌上限
tbf = mytbf_init(CPS,BURST);
if(tbf == NULL)
{
fprintf(stderr,"mytbf_init() failed.\n");
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\n",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 [email protected] $(CFLAGS) $(LDFLAGS)
clean:
rm -rf *.o mytbf
时间： 2024-11-02 12:49:39