- 論壇徽章:
- 0
|
用條件變量和鎖進(jìn)行線程控制，問題如注釋所寫。就是消費(fèi)者線程內(nèi)的那個(gè)usleep語句，為什么它那么重要
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
//多個(gè)生產(chǎn)者多個(gè)消費(fèi)者
#define BUFFSIZ 100000
#define MAXTHREADS 10
struct
{
int value;
pthread_t tid;
} buf[BUFFSIZ];
struct
{
pthread_mutex_t mutex;
int index;//生產(chǎn)者使用，索引buf位置
} shared = { PTHREAD_MUTEX_INITIALIZER, 0};
struct
{
pthread_mutex_t mutex;
pthread_cond_t cond;
int ready;//在該索引及之前的元素可以消費(fèi)
int consume;//消費(fèi)者將要消費(fèi)的元素下標(biāo)
} nready = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, -1, 0};
void *produce(void *arg)
{
for (;;)
{
pthread_mutex_lock(&shared.mutex);
if (shared.index > BUFFSIZ - 1)
{
pthread_mutex_unlock(&shared.mutex);
return NULL;
}
buf[shared.index].value = shared.index;
buf[shared.index].tid = pthread_self();
++shared.index;
pthread_mutex_unlock(&shared.mutex);
pthread_mutex_lock(&nready.mutex);
++nready.ready;
if ( nready.ready >= nready.consume)
pthread_cond_broadcast(&nready.cond);
pthread_mutex_unlock(&nready.mutex);
++(*(int *)arg);
usleep(1);
}
}
void *consume(void *arg)
{
int i;
for (;;)
{
pthread_mutex_lock(&nready.mutex);
if ( nready.consume>= BUFFSIZ)
{
pthread_mutex_unlock(&nready.mutex);
// printf("consume thread %lu is donen", pthread_self());
return NULL;
}
while ( nready.ready < nready.consume)
pthread_cond_wait(&nready.cond, &nready.mutex);
if ( buf[nready.consume].value != nready.consume)
printf("error: buf[%d]=%d, tid = %lun", nready.consume, buf[nready.consume].value, buf[nready.consume].tid);
++nready.consume;
pthread_mutex_unlock(&nready.mutex);
usleep(1);// 如果沒有這句，多個(gè)消費(fèi)者就會死鎖。
++(*(int *)arg);
}
}
int main(int argc, char *argv[])
{
int i, nprod, ncons ;
pthread_t tid_produce[MAXTHREADS], tid_consume[MAXTHREADS];
int pcount[MAXTHREADS] = {0};
int ccount[MAXTHREADS] = {0};
if (argc != 3)
perror("usage: a.out <#producer線程數(shù)> <#consumer線程數(shù)>"), exit(1);
nprod = atoi(argv[1])> MAXTHREADS ? MAXTHREADS : atoi(argv[1]);
ncons = atoi(argv[2])> MAXTHREADS ? MAXTHREADS : atoi(argv[2]);
pthread_setconcurrency(nprod+ncons);
for (i = 0; i < nprod; i++)
pthread_create(&tid_produce[i], NULL, produce, &pcount[i]);
for (i = 0; i < ncons; i++)
pthread_create(&tid_consume[i], NULL, consume, &ccount[i]);
for (i = 0; i < nprod; i++)
{
pthread_join(tid_produce[i], NULL);
printf("produce thread %lu is donen", tid_produce[i]);
}
for (i = 0; i < ncons; i++)
{
pthread_join(tid_consume[i], NULL);
printf("consume thread %lu is donen", tid_consume[i]);
}
for(i=0; i<nprod; ++i)
printf(" producer %d n", pcount[i]);
for(i=0; i<ncons; ++i)
printf(" consumer %d n", ccount[i]);
exit(0);
}
|
|
|
免費(fèi)注冊
發(fā)表于 2009-11-13 20:44
