source: src/tests/concurrent/examples/boundedBuffer.c@ 59073fd

//
// The contents of this file are covered under the licence agreement in the
// file "LICENCE" distributed with Cforall.
//
// boundedBuffer.c --
//
// Author           : Peter A. Buhr
// Created On       : Mon Oct 30 12:45:13 2017
// Last Modified By : Peter A. Buhr
// Last Modified On : Wed Dec 13 21:01:49 2017
// Update Count     : 27
//

#include <stdlib>
#include <fstream>                                                                              // random
#include <kernel>
#include <thread>
#include <unistd.h>                                                                             // getpid

forall( otype T )
monitor Buffer {
        condition full, empty;
        int front, back, count;
        T elements[20];
};

forall( otype T )
void ?{}( Buffer(T) & buffer ) {
        buffer.front = buffer.back = buffer.count = 0;
}

forall( otype T )
int query( Buffer(T) & buffer ) { return buffer.count; }

forall( otype T )
void insert( Buffer(T) & mutex buffer, T elem ) with( buffer ) {
        if ( count == 20 ) wait( empty );
        elements[back] = elem;
        back = ( back + 1 ) % 20;
        count += 1;
        signal( full );
}

forall( otype T )
T remove( Buffer(T) & mutex buffer ) with( buffer ) {
        if ( count == 0 ) wait( full );
        T elem = elements[front];
        front = ( front + 1 ) % 20;
        count -= 1;
        signal( empty );
        return elem;
}

thread Producer {
        Buffer(int) & buffer;
        unsigned int N;
};
void main( Producer & prod ) {
        for ( int i = 1; i <= prod.N; i += 1 ) {
                yield( random( 5 ) );
                insert( prod.buffer, 1 );
        } // for
        insert( prod.buffer, -1 );
}
void ?{}( Producer & prod, Buffer(int) * buffer, unsigned int N ) {
        &prod.buffer = buffer;
        prod.N = N;
}

thread Consumer {
        Buffer(int) & buffer;
        int & sum;                                              // summation of producer values
};
void main( Consumer & cons ) {
        cons.sum = 0;
        for ( ;; ) {
                yield( random( 5 ) );
                int item = remove( cons.buffer );
                if ( item == -1 ) break;                                // sentinel ?
                cons.sum += item;
        } // for
}
void ?{}( Consumer & cons, Buffer(int) * buffer, int * sum ) {
        &cons.buffer = buffer;
        &cons.sum = sum;
}

int main() {
        Buffer(int) buffer;
        enum { Prods = 5, Cons = 5 };
        Producer * prods[Prods];
        Consumer * cons[Cons];
        const int Sentinel = -1;
        int sums[Cons];
        int i;
        processor p;

        //random_seed( getpid() );
        random_seed( 1003 );

        for ( i = 0; i < Cons; i += 1 ) {                       // create consumers
                cons[i] = new( &buffer, &sums[i] );
        } // for
        for ( i = 0; i < Prods; i += 1 ) {                      // create producers
                prods[i] = new( &buffer, 100000u );
        } // for

        for ( i = 0; i < Prods; i += 1 ) {                      // wait for producers to finish
                delete( prods[i] );
        } // for
        for ( i = 0; i < Cons; i += 1 ) {                       // generate sentinal values to stop consumers
                insert( buffer, Sentinel );
        } // for
        int sum = 0;
        for ( i = 0; i < Cons; i += 1 ) {                       // wait for consumers to finish
                delete( cons[i] );
                sum += sums[i];
        } // for
        sout | "total:" | sum | endl;
}

// Local Variables: //
// tab-width: 4 //
// compile-command: "cfa boundedBuffer.c" //
// End: //