source: tests/concurrent/channels/chan_big.cfa @ 4a962d8

#include <locks.hfa>
#include <fstream.hfa>
#include <stdio.h>
#include <string.h>
#include <channel.hfa>
#include <thread.hfa>
#include <time.hfa>
#include <stats.hfa>
size_t Processors = 10, Channels = 10, Producers = 40, Consumers = 40, ChannelSize = 128;

owner_lock o;

unsigned long long total_operations = 0;

#define BIG 1

struct bigObject {
    size_t a;
    size_t b;
    size_t c;
    size_t d;
    size_t e;
    size_t f;
    size_t g;
    size_t h;
};

void ?{}( bigObject & this, size_t i ) with(this) { a = i; b = i; c = i; d = i; e = i; f = i; g = i; h = i; }
void ?{}( bigObject & this ) { this{0}; }

#ifdef BIG
typedef channel( bigObject ) Channel;
#else
typedef channel Channel;
#endif

Channel * channels;

volatile bool cons_done = false, prod_done = false;
size_t cons_check = 0, prod_check = 0;

thread Consumer {
    size_t i;
};
static inline void ?{}( Consumer & c, size_t i, cluster & clu ) {
    ((thread &)c){ clu };
    c.i = i; 
}
void main(Consumer & this) {
    unsigned long long runs = 0;
    size_t my_check = 0;
    for ( ;; ) {
        if ( cons_done ) break;
        #ifdef BIG
        bigObject j = remove( channels[ this.i ] );
        my_check = my_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
        #else
        size_t j = remove( channels[ this.i ] );
        my_check = my_check ^ j;
        #endif
        
        if ( !prod_done ) runs++;
    }
    lock(o);
    total_operations += runs;
    cons_check = cons_check ^ my_check;
    // sout | "C: " | runs;
    unlock(o);
}

thread Producer {
    size_t i;
};
static inline void ?{}( Producer & p, size_t i, cluster & clu ) {
    ((thread &)p){ clu };
    p.i = i;
}
void main(Producer & this) {
    unsigned long long runs = 0;
    size_t my_check = 0;
    for ( ;; ) {
        if ( prod_done ) break;
        #ifdef BIG
        bigObject j{(size_t)runs};
        insert( channels[ this.i ], j );
        my_check = my_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
        #else
        insert( channels[ this.i ], (size_t)runs );
        my_check = my_check ^ ((size_t)runs);
        #endif
        runs++;
    }
    lock(o);
    total_operations += runs;
    prod_check = prod_check ^ my_check;
    // sout | "P: " | runs;
    unlock(o);
}


int main( int argc, char *argv[] ) {
    switch( argc ) {
      case 6:
                if ( strcmp( argv[5], "d" ) != 0 ) {                    // default ?
                        if ( atoi( argv[5] ) < 1) goto Usage;
                        ChannelSize = atoi( argv[5] );
                } // if
      case 5:
                if ( strcmp( argv[4], "d" ) != 0 ) {                    // default ?
                        if ( atoi( argv[4] ) < 1 ) goto Usage;
                        Channels = atoi( argv[4] );
                } // if
      case 4:
                if ( strcmp( argv[3], "d" ) != 0 ) {                    // default ?
                        if ( atoi( argv[3] ) < 1 ) goto Usage;
                        Consumers = atoi( argv[3] );
                } // if
      case 3:
                if ( strcmp( argv[2], "d" ) != 0 ) {                    // default ?
                        if ( atoi( argv[2] ) < 1 ) goto Usage;
                        Producers = atoi( argv[2] );
                } // if
      case 2:
                if ( strcmp( argv[1], "d" ) != 0 ) {                    // default ?
                        if ( atoi( argv[1] ) < 1 ) goto Usage;
                        Processors = atoi( argv[1] );
                } // if
          case 1:                                                                                       // use defaults
                break;
          default:
          Usage:
                sout | "Usage: " | argv[0]
             | " [ processors > 0 | d ]"
             | " [ ProdsPerChan > 0 | d ]"
             | " [ ConsPerChan > 0 | d ]"
             | " [ channels > 0 | d ]";
                exit( EXIT_FAILURE );
    }

    size_t Clusters = 1;
    // create a cluster
    cluster clus[Clusters];
    processor * proc[Processors];
    for ( i; Processors ) {
        (*(proc[i] = alloc())){clus[i % Clusters]};
    }

    channels = anew( Channels );

    // sout | "Processors: " | Processors | " ProdsPerChan: " | Producers | " ConsPerChan: " | Consumers | "Channels: " | Channels | " Channel Size: " | ChannelSize;
    
    for ( i; Channels ) {
        channels[i]{ ChannelSize };
    }

    sout | "start";
    Consumer * c[Consumers * Channels];
    Producer * p[Producers * Channels];

    for ( i; Consumers * Channels ) {
        (*(c[i] = alloc())){ i % Channels, clus[i % Clusters] };
    }

    for ( i; Producers * Channels ) {
        (*(p[i] = alloc())){ i % Channels, clus[i % Clusters] };
    }

    sleep(10`s);
    prod_done = true;

    for ( i; Producers * Channels ) {
        delete(p[i]);
    }

    sout | "prods";
    cons_done = true;
    for ( i; Channels ) {
        // sout | get_count( channels[i] );
        if ( get_count( channels[i] ) < Consumers ){
            #ifdef BIG
            bigObject b{0};
            #endif
            for ( j; Consumers ) {
                #ifdef BIG
                insert( channels[i], b );
                #else
                insert( channels[i], 0 );
                #endif
            }
        }
    }
    sout | "cons";
    for ( i; Consumers * Channels ) {
        delete(c[i]);
    }

    sout | "flush";
    for ( i; Channels ) {
        for ( ;; ) {
            if ( get_count( channels[i] ) > 0 ) {
                #ifdef BIG
                bigObject j = remove( channels[ i ] );
                cons_check = cons_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
                #else
                size_t j = remove( channels[ i ] );
                cons_check = cons_check ^ j;
                #endif
            } else break;
        }
    }

    adelete( channels );
    // sout | "total channel ops: " | total_operations;
    if ( cons_check != prod_check )
        sout | "CHECKSUM MISMATCH !!!";
    // print_stats_now( *active_cluster(), CFA_STATS_READY_Q);

    for ( i; Processors ) {
        delete(proc[i]);
    }
    sout | "done";
    return 0;
}