source: tests/concurrency/examples/boundedBufferTHREAD.cfa@ efe697d

//
// Cforall Version 1.0.0 Copyright (C) 2018 University of Waterloo
// 
// The contents of this file are covered under the licence agreement in the
// file "LICENCE" distributed with Cforall.
//
// boundedBufferTHREAD.c --
//
// Author           : Peter A. Buhr
// Created On       : Wed Apr 18 22:52:12 2018
// Last Modified By : Peter A. Buhr
// Last Modified On : Thu Jan 16 23:09:43 2020
// Update Count     : 25
//

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

//Duration default_preemption() { return 0; }

enum { BufferSize = 50 };

//forall( otype T ) {
        thread Buffer {
                int front, back, count;
                int elements[BufferSize];
        }; // Buffer

        void ?{}( Buffer & buffer ) with( buffer ) { [front, back, count] = 0; }

        int query( Buffer & buffer ) { return buffer.count; } // read-only, no mutual exclusion

        void insert( Buffer & mutex buffer, int elem ) with( buffer ) {
                elements[back] = elem;
        } // insert

        int remove( Buffer & mutex buffer ) with( buffer ) {
                return elements[front];
        } // remove

        void main( Buffer & buffer ) with( buffer ) {
                for () {
                        waitfor( ^?{} : buffer ) {
                                break;
                        } or when ( count != 20 ) waitfor( insert : buffer ) {
                                back = (back + 1) % 20;
                                count += 1;
                        } or when ( count != 0 ) waitfor( remove : buffer ) {
                                front = (front + 1) % 20;
                                count -= 1;
                        } // waitfor
                } // for
        } // main
//}

enum { Sentinel = -1 };

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

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

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

        //srandom( getpid() );
        srandom( 1003 );

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

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

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