source: tests/concurrency/pthread/bounded_buffer.cfa @ f6fd22a4

Last change on this file since f6fd22a4 was c26bea2a, checked in by Peter A. Buhr <pabuhr@…>, 19 months ago

first attempt at renaming directory tests/concurrent to tests/concurrency to harmonize with other concurrency directory names

  • Property mode set to 100644
File size: 5.1 KB
Line 
1#include <stdlib.h>                                                                             // prototype: rand
2#include <fstream.hfa>
3#include <thread.hfa>
4#include <pthread.h>
5#include <errno.h>
6// tested pthread mutex related routines, pthread cond related routines
7// tested pthread_create/join
8
9enum { BufferSize = 50 };
10
11volatile int producer_val_total;
12volatile int consumer_val_total;
13
14pthread_mutex_t producer_cnt_lock, consumer_cnt_lock;
15
16
17forall( T ){
18    struct Buffer
19    {
20        int front, back, count;
21                T elements[BufferSize];
22        pthread_mutex_t _mutex;
23            pthread_cond_t Full, Empty;                                                 // waiting consumers & producers
24    };
25
26    void ?{}( Buffer(T) & buffer ) with( buffer ) {
27         [front, back, count] = 0;
28         pthread_mutex_init(&_mutex, NULL);
29         pthread_cond_init(&Full, NULL);
30         pthread_cond_init(&Empty, NULL);
31    }
32
33    void ^?{}( Buffer(T) & buffer ) with( buffer ){
34        pthread_mutex_destroy(&_mutex);
35        pthread_cond_destroy(&Full);
36        pthread_cond_destroy(&Empty);
37    }
38
39    int query( Buffer(T) & buffer ) { return buffer.count; } // read-only, no mutual exclusion
40   
41    void insert( Buffer(T) & buffer, T elem ) with(buffer) {
42                pthread_mutex_lock(&_mutex);
43                while ( count == 20 ) pthread_cond_wait( &Empty, &_mutex ); // block producer
44                elements[back] = elem;
45                back = ( back + 1 ) % 20;
46                count += 1;
47                pthread_cond_signal( &Full );                                   // unblock consumer
48        pthread_mutex_unlock(&_mutex);
49        }
50
51    T remove(Buffer(T) & buffer) with(buffer) {
52                pthread_mutex_lock(&_mutex);
53                while ( count == 0 ) pthread_cond_wait( &Full, &_mutex ); // block consumer
54                T elem = elements[front];
55                front = ( front + 1 ) % 20;
56                count -= 1;
57                pthread_cond_signal( &Empty );                                  // unblock producer
58        pthread_mutex_unlock(&_mutex);
59                return elem;
60        }
61
62}
63
64void *producer( void *arg ) {
65    Buffer(int) &buf = *(Buffer(int)*)arg;
66        const int NoOfItems = prng(*active_thread(), 40);
67        int item;
68        for ( int i = 1; i <= NoOfItems; i += 1 ) {                     // produce a bunch of items
69                item = prng(*active_thread(), 1, 101);                  // produce a random number
70                //sout | "Producer:" | pthread_self() | " value:" | item;
71                insert( buf,item );                                                             // insert element into queue
72        pthread_mutex_lock(&producer_cnt_lock);
73        producer_val_total += item;
74        pthread_mutex_unlock(&producer_cnt_lock);
75        } // for
76        //sout | "Producer:" | pthread_self() | " is finished";
77        return NULL;
78} // producer
79
80void *consumer( void *arg ) {
81        Buffer(int) &buf = *(Buffer(int) *)arg;
82        int item;
83        for ( ;; ) {                                                                            // consume until a negative element appears
84                item = remove(buf);                                                     // remove from front of queue
85                //sout | "Consumer:" | pthread_self() | " value:" | item;
86          if ( item == -1 ) break;
87        pthread_mutex_lock(&consumer_cnt_lock);
88        consumer_val_total += item;
89        pthread_mutex_unlock(&consumer_cnt_lock);
90        } // for
91        //sout | "Consumer:" | pthread_self() | " is finished";
92        return NULL;
93} // consumer
94
95int main() {
96        const int NoOfCons = 20, NoOfProds = 30;
97        Buffer(int) buf;                                                                // create a buffer monitor
98        pthread_t cons[NoOfCons];                                                       // pointer to an array of consumers
99        pthread_t prods[NoOfProds];                                                     // pointer to an array of producers
100    pthread_mutex_init(&producer_cnt_lock, NULL);
101    pthread_mutex_init(&consumer_cnt_lock, NULL);
102        // parallelism
103    set_seed( 1003 );
104
105        processor p[5];
106    {
107        // create/join and mutex/condition test
108        //sout | "create/join and mutex/condition test";
109        for ( int i = 0; i < NoOfCons; i += 1 ) {                       // create consumers
110            if ( pthread_create( &cons[i], NULL, consumer, (void*)&buf ) != 0 ) {
111                sout | "create thread failure, errno:" | errno;
112                exit( EXIT_FAILURE );
113            } // if
114        } // for
115        for ( int i = 0; i < NoOfProds; i += 1 ) {                      //      create producers
116            if ( pthread_create( &prods[i], NULL, producer, (void*)&buf ) != 0 ) {
117                sout | "create thread failure";
118                exit( EXIT_FAILURE );
119            } // if
120        } // for
121
122        void *result;
123        for ( int i = 0; i < NoOfProds; i += 1 ) {                      // wait for producers to end
124            if ( pthread_join( prods[i], &result ) != 0 ) {
125                sout | " producers join thread failure";
126                exit( EXIT_FAILURE );
127            } // if
128            if ( (uint64_t)result != 0 ) {
129                sout | "producers" | prods[i] |" bad return value " | result;
130                exit( EXIT_FAILURE );
131            } // if
132            //sout | "join prods[" | i | "]:" | prods[i] | " result:" | result;
133        } // for
134
135        for ( int i = 0; i < NoOfCons; i += 1 ) {                       // terminate each consumer
136            insert(buf, -1 );
137        } // for
138
139        for ( int i = 0; i < NoOfCons; i += 1 ) {                       // wait for consumer to end
140            if ( pthread_join( cons[i], &result ) != 0 ) {
141                sout| "consumers join thread failure" ;
142                exit( EXIT_FAILURE );
143            } // if
144            if ( (uint64_t)result != 0 ) {
145                sout| "consumers bad return value" | result;
146                exit( EXIT_FAILURE );
147            } // if
148        } // for
149        sout | "producer total value is " | producer_val_total;
150        sout | "consumer total value is " | consumer_val_total;
151    }
152
153       
154
155       
156}
Note: See TracBrowser for help on using the repository browser.