1/* 2 * Copyright (c) 2002, Intel Corporation. All rights reserved. 3 * Created by: crystal.xiong REMOVE-THIS AT intel DOT com 4 * This file is licensed under the GPL license. For the full content 5 * of this license, see the COPYING file at the top level of this 6 * source tree. 7 * 8 * This is a test about multiple producers and consumers. Producers send data 9 * to a buffer. Consumers keeps reading data from the buffer. 10 */ 11 12#include <stdio.h> 13#include <unistd.h> 14#include <fcntl.h> 15#include <stdlib.h> 16#include <sys/wait.h> 17#include <sys/mman.h> 18#include <string.h> 19#include <getopt.h> 20#include <errno.h> 21#include <pthread.h> 22 23#include <semaphore.h> 24#include "posixtest.h" 25 26#define BUF_SIZE 5 27#define Max_Num 5 28#define Max_Threads 127 29 30typedef struct { 31 int buffer[BUF_SIZE]; 32 sem_t full; 33 sem_t empty; 34 sem_t lock; 35}buf_t; 36 37buf_t *buf; 38int in, out; 39 40int *producer(void *ID) 41{ 42 int data; 43 int i; 44 int ThreadID = *(int *)ID; 45 int full_value; 46 47 printf("Enter into Producer Thread %d... \n", ThreadID); 48 for (i = 0; i< Max_Num - 1; i++) { 49 if (-1 == sem_wait(&buf->full)) { 50 perror("sem_wait didn't return success \n"); 51 pthread_exit((void *)1); 52 } 53 if (-1 == sem_getvalue(&buf->full, &full_value)) { 54 perror("sem_getvalue didn't return success \n"); 55 pthread_exit((void *)1); 56 } 57 printf("The value of the full semaphore is %d \n", full_value); 58 if (-1 == sem_wait(&buf->lock)) { 59 perror("sem_wait didn't return success \n"); 60 pthread_exit((void *)1); 61 } 62 data = 100*ThreadID + i; 63 buf->buffer[in] = data; 64 printf("[%d] producer has added %d to the buffer[%d] \n", ThreadID, data, in); 65 in = (in + 1) % BUF_SIZE; 66 if (-1 == sem_post(&buf->lock)) { 67 perror("sem_wait didn't return success \n"); 68 pthread_exit((void *)1); 69 } 70 if (-1 == sem_post(&buf->empty)) { 71 perror("sem_wait didn't return success \n"); 72 pthread_exit((void *)1); 73 } 74 } 75 if (-1 == sem_wait(&buf->full)) { 76 perror("sem_wait didn't return success \n"); 77 pthread_exit((void *)1); 78 } 79 if (-1 == sem_getvalue(&buf->full, &full_value)) { 80 perror("sem_getvalue didn't return success \n"); 81 pthread_exit((void *)1); 82 } 83 printf("The value of the full is %d \n", full_value); 84 if (-1 == sem_wait(&buf->lock)) { 85 perror("sem_wait didn't return success \n"); 86 pthread_exit((void *)1); 87 } 88 data = -1; 89 buf->buffer[in] = data; 90 printf("[%d] producer has added %d to the buffer[%d] \n", ThreadID, data, in); 91 in = (in + 1) % BUF_SIZE; 92 if (-1 == sem_post(&buf->lock)) { 93 perror("sem_wait didn't return success \n"); 94 pthread_exit((void *)1); 95 } 96 if (-1 == sem_post(&buf->empty)) { 97 perror("sem_wait didn't return success \n"); 98 pthread_exit((void *)1); 99 } 100 printf("Producer %d exit... \n", ThreadID); 101 pthread_exit((void *)0); 102} 103int *consumer(void *ID) 104{ 105 int data; 106 int ThreadID = *(int *)ID; 107 int full_value; 108 109 printf("Enter into Consumer Thread %d... \n", ThreadID); 110 do 111 { 112 if (-1 == sem_wait(&buf->empty)) { 113 perror("sem_wait didn't return success \n"); 114 pthread_exit((void *)1); 115 } 116 if (-1 == sem_wait(&buf->lock)) { 117 perror("sem_wait didn't return success \n"); 118 pthread_exit((void *)1); 119 } 120 data = buf->buffer[out]; 121 printf("[%d] consumer has taken %d from buffer[%d] \n", ThreadID, data, out); 122 out = (out + 1) % BUF_SIZE; 123 if (-1 == sem_post(&buf->lock)) { 124 perror("sem_wait didn't return success \n"); 125 pthread_exit((void *)1); 126 } 127 if (-1 == sem_post(&buf->full)) { 128 perror("sem_wait didn't return success \n"); 129 pthread_exit((void *)1); 130 } 131 if (-1 == sem_getvalue(&buf->full, &full_value)) { 132 perror("sem_getvalue didn't return success \n"); 133 pthread_exit((void *)1); 134 } 135 printf("The value of the full semaphore is %d \n", full_value); 136 } 137 while (data != -1); 138 139 printf("Consumer %d exit... \n", ThreadID); 140 pthread_exit((void *)0); 141} 142int main(int argc, char *argv[]) 143{ 144 int shared = 1; 145 int full_value = BUF_SIZE; 146 int empty_value = 0; 147 int lock_value=1; 148 int num; 149 int i; 150 pthread_t con[Max_Threads], pro[Max_Threads]; 151 int ThreadID[Max_Threads]; 152 153 154#ifndef _POSIX_SEMAPHORES 155 printf("_POSIX_SEMAPHORES is not defined \n"); 156 return PTS_UNRESOLVED; 157#endif 158 159 buf = (buf_t *)malloc(sizeof(buf_t)); 160 161 if ( (2 != argc) || (( num = atoi(argv[1])) <= 0)) { 162 fprintf(stderr, "Usage: %s number_of_threads\n", argv[0]); 163 return PTS_FAIL; 164 } 165 if (num > Max_Threads) { 166 printf("The num of producers/consumers threads are too large. Reset to %d\n", Max_Threads); 167 num = Max_Threads; 168 } 169 170 if (-1 == sem_init(&buf->full, shared, full_value)) { 171 perror("sem_init didn't return success \n"); 172 return PTS_UNRESOLVED; 173 } 174 if (-1 == sem_getvalue(&buf->full, &full_value)) { 175 perror("sem_getvalue didn't return success \n"); 176 return PTS_UNRESOLVED; 177 } 178 printf("The initial value of the full semaphore is %d \n", full_value); 179 if (-1 == sem_init(&buf->empty, shared, empty_value)) { 180 perror("sem_init didn't return success \n"); 181 return PTS_UNRESOLVED; 182 } 183 if (-1 == sem_init(&buf->lock, shared, lock_value)) { 184 perror("sem_init didn't return success \n"); 185 return PTS_UNRESOLVED; 186 } 187 in = out = 0; 188 189 for (i = 0; i < num; i++) { 190 ThreadID[i] = i; 191 pthread_create(&con[i], NULL, (void *)consumer, (void *)&ThreadID[i]); 192 } 193 for (i = 0; i < num; i++) { 194 ThreadID[i] = i; 195 pthread_create(&pro[i], NULL, (void *)producer, (void *)&ThreadID[i]); 196 } 197 for (i = 0; i < num; i++) 198 pthread_join(con[i], NULL); 199 for (i = 0; i < num; i++) 200 pthread_join(pro[i], NULL); 201 202 sem_destroy(&buf->full); 203 sem_destroy(&buf->empty); 204 sem_destroy(&buf->lock); 205 return PTS_PASS; 206} 207