文档章节

linux producer consumer sources

romalin99
 romalin99
发布于 2013/07/07 14:40
字数 1178
阅读 64
收藏 0

除了提供互斥之外,信号量的另外一个作用是调度对共享资源的访问。在这种情景中,一个线程用信号量来通知另一个线程,程序状态中的某个条件已经为真了。

下图给出了生产者-------消费者问题。生产者和消费者线程共享一个有n个槽的有限缓冲区。

生产者线程反复的生成新的项目(item),并把它们插入到缓冲区中。消费者线程不断的从缓冲区中取出这些项目,然后消费( 使用它们)
 

 

因为插入和取出项目都涉及更新共享变量,所以我们必须保证对缓冲区的访问时互斥的。

但是只保证互斥访问还是不够的,我们还需要调度  对缓冲区的访问 (如果缓冲区是满的即没有可用的槽,那么生产者必须等待直到有一个空的槽变为可用为止。与之相似,若缓冲区是空的即没有可取用的槽, 那么消费者必须等待直到有一个项目变为可用。

接下来我们为生产者----消费者  定义一个结构体: 来存储数据

Typedef struct {

       Int * buf;      //  item存放在一个动态分配的n项整数buf

       Int n;         //

       Int front;     //  索引值记录第一项 和最后一项

       Int rear;     //

       Sem_t mutex;  //  三个信号量同步对缓冲区的访问。提供互斥的缓冲区的访问。

       Sem_t slots; //   分别记录空槽和可用item的数量

       Sem_t items; //

}sbuf_t;

/*

Buffer array

Maximum number of slots

Buf[(front+1)%n]  is first item

Buf[rear%n] is last item

Protects accessed to buf

Counts available slots

Counts available items

*/

此结构体包含使用的有限缓冲区

我们使用一个函数sbuf_init来初始化此缓冲区,并设置front rear 表示一个空的缓冲区,并为三个信号量赋予初始值。使用sbuf_deinit函数来删除缓冲区(当程序使用完之后)

Sbuf_insert函数等待一个可用的槽,对互斥锁加锁,添加项目item。对互斥锁解锁,然后宣布有一个新的item可用。

Sbuf_remove函数式与上一个函数对应的。在等待一个可用的缓冲区之后,对互斥锁加锁,从缓冲区的前面取出该项目,对互斥锁解锁。然后发信号通知一个新的槽可供使用。

 

 

void  sbuf_init(sbuf_t *sp,int n)

{

       If((Sp->buf=calloc(n,sizeof(int))==NULL)

              Printf(“calloc  错误\n”);

       Sp->n=n; //Buffer holds max of n items

       Sp->front=sp->rear=0;  //Empty buffer iff front==rear

       Sem_init(&sp->mutex,0,1);//binary semaphore for locking

       Sem_init(&sp->slots,0,n); //Initially ,buf has n empty slots

       Sem_init(&sp->items,0,0); //initially, buf has zero data items

}

Void sbuf_deinit(sbuf_t *sp)

{

       Free(sp->buf);

}

Void sbuf_insert(sbuf_t *sp,int item)

{

       Sem_wait(&sp->slots);//wait for available slot

       Sem_wait(&sp->mutex);//lock the buffer

       Sp->buf[(++sp->rear)%(sp->n)]=item;//insert the item

       Sem_post(&sp->mutex);//unlock the buffer

       Sem_post(&sp->items);//announce available item

}

 

Int sbuf_remove(sbuf_t *sp)

{

       Int item;

       Sem_wait(&sp->items);//wait for available item

       Sem_wait(&sp->mutex);//lock the buffer

       Item=sp->buf[(++sp->front)%(sp->n)];// remove the item

       Sem_post(&sp->mutex);

       Sem_post(&sp->slots);

       Return item;

}

#include  <stdio.h>

#include  <stdlib.h>
#include  <string.h>
#include  <pthread.h>
#include  <unistd.h>
#include  <semaphore.h>

typedef struct {
    int * buf;      //  item存放在一个动态分配的n项整数buf中
    int n;         //
    int front;     //  索引值记录第一项 和最后一项
    int rear;     //
    sem_t mutex;  //  三个信号量同步对缓冲区的访问。提供互斥的缓冲区的访问。
    sem_t slots; //   分别记录空槽和可用item的数量
    sem_t items; //
}sbuf_t;
sbuf_t g_sp;
void  sbuf_init(sbuf_t *sp,int n)
{
    sp->buf=(int*)calloc(n,sizeof(int));
    if( NULL == sp->buf)
        printf("calloc  错误\n");
    sp->n=n; //Buffer holds max of n items
    sp->front=sp->rear=0;  //Empty buffer iff front==rear
    sem_init(&sp->mutex,0,1);//binary semaphore for locking
    sem_init(&sp->slots,0,n); //Initially ,buf has n empty slots
    sem_init(&sp->items,0,0); //initially, buf has zero data items
}
void sbuf_deinit(sbuf_t *sp)
{
    free(sp->buf);
}
void sbuf_insert(sbuf_t *sp,int item)
{
    sem_wait(&sp->slots);//wait for available slot
    sem_wait(&sp->mutex);//lock the buffer
    sp->buf[(++sp->rear)%(sp->n)]=item;//insert the item
    sem_post(&sp->mutex);//unlock the buffer
    sem_post(&sp->items);//announce available item
}

int sbuf_remove(sbuf_t *sp)
{
    int item;
    sem_wait(&sp->items);//wait for available item
    sem_wait(&sp->mutex);//lock the buffer
    item=sp->buf[(++sp->front)%(sp->n)];// remove the item
    sem_post(&sp->mutex);
    sem_post(&sp->slots);
    return item;
}

void* producer(void* arg)
{
    int i,item,index;
    //index 是线程编号
    index=*(int*)arg;
    for(int i=0;i<g_sp.n;i++)
    {
        item =(i*2+1);
        sbuf_insert(&g_sp,item);
        printf("[P%d] Producing %d ...\n",index,item);
        fflush(stdout);
        sleep(1);
    }
}
void* consumer(void* arg)
{
    int i,item,index;
    index=*(int*)arg;
    for(int i=0;i<g_sp.n;i++)
    {
        item =sbuf_remove(&g_sp);
        printf("------>[C%d] Consuming %d ...\n",index,item);
        sleep(1);
    }
}

int main(int argc,char* argv[])
{
    sbuf_init(&g_sp,10);
    int NP=3,NC=3;
    pthread_t idp[NP],idc[NC];
    //1
    for(int i=0;i<NP;i++)
    {
            pthread_create(&idp[i],NULL,producer,(void*)&i);
    }
    //2
    for(int i=0;i<NC;i++)
    {
            pthread_create(&idc[i],NULL,consumer,(void*)&i);
    }
    //3
    for(int i=0;i<NP;i++)
        pthread_join(idp[i],NULL);
    for(int i=0;i<NC;i++)
        pthread_join(idc[i],NULL);
        
    sbuf_deinit(&g_sp);
    return 0;
}

© 著作权归作者所有

共有 人打赏支持
romalin99
粉丝 8
博文 153
码字总数 76769
作品 0
浦东
高级程序员
私信 提问
circular buffer in Linux kernel

/ See Documentation/circular-buffers.txt for more information. */ #ifndef LINUXCIRCBUFH #define LINUXCIRCBUFH 1 struct circ_buf { char *buf; int head; int tail; }; / Return coun......

ChenQi
2012/03/23
0
0
Using Kafka with Flume

这个文档是 Cloudera Distribution of Apache Kafka 1.3.x. 其他版本的文档在Cloudera Documentation. Using Kafka with Flume 在CDH 5.2.0 及更高的版本中, Flume 包含一个Kafka source an......

晨磊
2015/08/29
852
0
Apache Samza 0.14.0 发布,分布式流处理框架

Apache Samza 是一个分布式流处理框架,专用于实时数据的处理,目前已经在几个大公司(包括 LinkedIn、Netflix、Uber、Slack、Redfin、TripAdvisor)实际生产中使用。 0.14.0 版本包含以下备...

王练
2018/01/07
590
0
协程技术研究

先记住一句话,子程序是协程的一种特例。 Python yield Python 中的 yield 保存一个 generator 函数的状态,generator 是一个特殊类型的迭代器(iterator) 运行结果 yield 相当于 return,主...

兔之
2016/11/24
114
0
Kafka部署

简介 kafka (官网地址:http://kafka.apache.org)是一款分布式消息发布和订阅的系统,具有高性能和高吞吐率。 i. 消息的发布(publish)称作producer,消息的订阅(subscribe)称作consume...

follitude
2016/09/21
17
0

没有更多内容

加载失败,请刷新页面

加载更多

关于php的xdebug配置(编辑器vscode)

虽然说echo和print_r是公认的最实用的调试工具,但是效率真的没有可视化的直接断点那么高。这里简单介绍如果安装及配置xdebug for vscode 一、PHP环境处的配置 1、编译安装 下载及编译php,因...

元谷
32分钟前
7
0
heartbeat

http://www.linux-ha.org/doc/users-guide/_building_and_installing_heartbeat_from_source.html kaer@linux-sqlf:~/Reusable-Cluster-Components-glue--0a7add1d9996> ./configure --enabl......

李有常
42分钟前
1
0
shell 脚本 title 免写入

1、有时候写脚本,需要写一些注释信息或者备注之类,当每次都手写比较麻烦,可以写一个脚本,每次运行则自动写入这些信息,不需要在手动输入; 通过 EOF 嵌入文档的方式; [root@localhost_002 sh...

芬野de博客
48分钟前
3
0
FragmentPagerAdapter+ViewPager 更新问题

场景 存在一种需求,当用户系统中,属于某一组织的用户登录之后(或者账户切换),要求主页面显示不同的ViewPager + Fragment组合,并且要求app无需退出就能刷新组合以及组合中的页面。 此外...

IamOkay
49分钟前
3
0
Python有哪些常用的标准库?Python基础教程,第十一讲,常用标准库(一)

上一节课我们学习了如何自定义模块,大家可以定义自己的模块,但是项目中通常不允许我们自己造轮子,所以本节课主要学习Python的标准库,学习使用别人的轮子,由于标准库内容较多,所以我们分...

程序员补给栈
57分钟前
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部