source: tests/concurrency/waituntil/channel_close.cfa @ ec22220

#include <fstream.hfa>
#include <thread.hfa>
#include <channel.hfa>
#include <time.hfa>

channel(ssize_t) A, B;

volatile size_t inserts = 0, removes = 0;

thread Producer {};
void main( Producer & this ) {
    try {
        for( size_t i; 0~@ ) {
            waituntil( A << i ) { inserts++; }
            and waituntil( B << i ) { inserts++; }
        }
    } catch ( channel_closed * e ) {} 
}

bool useAnd = false;
thread Consumer {}; // ensures that the changing when states of Server1 don't result in a deadlock
void main( Consumer & this ) {
    ssize_t in, in2, A_removes = 0, B_removes = 0;

    try {
        for () {
            if ( useAnd ) {
                waituntil( (in << A) ) { __atomic_thread_fence( __ATOMIC_SEQ_CST ); assert( A_removes == in ); A_removes++; removes++; }
                and waituntil( (in2 << B) ) { __atomic_thread_fence( __ATOMIC_SEQ_CST ); assert( B_removes == in2 ); B_removes++; removes++; }
                continue;
            }
            waituntil( (in << A) ) { __atomic_thread_fence( __ATOMIC_SEQ_CST ); assert( A_removes == in ); A_removes++; removes++; }
            or waituntil( (in << B) ) { __atomic_thread_fence( __ATOMIC_SEQ_CST ); ( B_removes == in ); B_removes++; removes++; }
        }
    } catchResume ( channel_closed * e ) {                              // continue to remove until would block
        } catch ( channel_closed * e ) {} 

    try {
        for ()
            waituntil( (in << A) ) { __atomic_thread_fence( __ATOMIC_SEQ_CST ); assert( A_removes == in ); A_removes++; removes++; }
    } catchResume ( channel_closed * e ) {                              // continue to remove until would block
        } catch ( channel_closed * e ) {} 

    try {
        for ()
            waituntil( (in << B) ) { __atomic_thread_fence( __ATOMIC_SEQ_CST ); assert( B_removes == in ); B_removes++; removes++; }
    } catchResume ( channel_closed * e ) {                              // continue to remove until would block
        } catch ( channel_closed * e ) {} 
}

int main( int argc, char * argv[] ) {
        size_t time = 5;

    if ( argc == 2 )
        time = atoi( argv[1] );

    processor p[2];
    A{5};
    B{5};

    sout | "start OR";
    {
        Producer p;
        Consumer c;
        sleep(time`s);
        sout | "done sleep";
        sout | "closing A";
        close( A );
        sout | "closing B";
        close( B );
    }
    if ( inserts != removes ) 
        sout | "CHECKSUM MISMATCH!! Producer got:" | inserts | ", Consumer got:" | removes;
    sout | "done";
    ^A{};
    ^B{};

    useAnd = true;
    inserts = removes = 0;
    A{5};
    B{5};
    sout | "start AND";
    {
        Producer p;
        Consumer c;
        sleep( time`s );
        sout | "done sleep";
        sout | "closing A";
        close( A );
        sout | "closing B";
        close( B );
    }
    if ( inserts != removes ) 
        sout | "CHECKSUM MISMATCH!! Producer got:" | inserts | ", Consumer got:" | removes;
    sout | "done";
}