Index: tests/concurrent/channels/.expect/churn.txt =================================================================== --- tests/concurrent/channels/.expect/churn.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/.expect/churn.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrent/channels/.expect/contend.txt =================================================================== --- tests/concurrent/channels/.expect/contend.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/.expect/contend.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrent/channels/.expect/daisy_chain.txt =================================================================== --- tests/concurrent/channels/.expect/daisy_chain.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/.expect/daisy_chain.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrent/channels/.expect/hot_potato.txt =================================================================== --- tests/concurrent/channels/.expect/hot_potato.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/.expect/hot_potato.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrent/channels/.expect/ping_pong.txt =================================================================== --- tests/concurrent/channels/.expect/ping_pong.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/.expect/ping_pong.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrent/channels/.expect/pub_sub.txt =================================================================== --- tests/concurrent/channels/.expect/pub_sub.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/.expect/pub_sub.txt (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrent/channels/barrier.cfa =================================================================== --- tests/concurrent/channels/barrier.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/barrier.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,130 @@ +#include +#include +#include +#include +#include +#include +#include + +size_t total_operations = 0; +int Processors = 1, Tasks = 5, BarrierSize = 2; + +typedef channel( int ) Channel; + +Channel * barWait; +Channel * entryWait; +owner_lock o; + +bool done = false; +size_t tasks_done = 0; + +static inline void initBarrier() { + for ( j; BarrierSize ) + insert( *entryWait, j ); +} + +static inline void barrier() { + int ticket = remove( *entryWait ); + + if ( ticket == BarrierSize - 1 ) { + for ( j; BarrierSize - 1 ) + insert( *barWait, j ); + return; + } + ticket = remove( *barWait ); + + // last one out + if ( BarrierSize == 1 || ticket == BarrierSize - 2 ) { + for ( j; BarrierSize ) + insert( *entryWait, j ); + } +} + +thread Task {}; +static inline void ?{}( Task & p, cluster & clu ) { + ((thread &)p){ clu }; +} +void main(Task & this) { + size_t runs = 0; + try { + for ( ;; ) { + if ( done ) break; + barrier(); + runs++; + } + } catch ( channel_closed * e ) {} + lock(o); + total_operations += runs; + // sout | "P: " | runs; + unlock(o); +} + + +int main( int argc, char * argv[] ) { + switch ( argc ) { + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + BarrierSize = atoi( argv[2] ); + if ( Processors < 1 ) goto Usage; + } // if + case 2: + if ( strcmp( argv[1], "d" ) != 0 ) { // default ? + Processors = atoi( argv[1] ); + if ( Processors < 1 ) goto Usage; + } // if + case 1: // use defaults + break; + default: + Usage: + sout | "Usage: " | argv[0] + | " [ processors (> 0) | 'd' (default " | Processors + | ") ] [ BarrierSize (> 0) | 'd' (default " | BarrierSize + | ") ]" ; + exit( EXIT_FAILURE ); + } // switch + if ( Tasks < BarrierSize ) + Tasks = BarrierSize; + + size_t Clusters = 1; + // create a cluster + cluster clus[Clusters]; + processor * proc[Processors]; + for ( i; Processors ) { + (*(proc[i] = malloc())){clus[i % Clusters]}; + } + + Channel entry{ BarrierSize }; + Channel wait{ BarrierSize }; + + entryWait = &entry; + barWait = &wait; + + initBarrier(); + + sout | "start"; + Task * t[Tasks]; + + for ( i; Tasks ) { + (*(t[i] = malloc())){ clus[i % Clusters] }; + } + + sleep(10`s); + done = true; + + sout | "sleep"; + + close( entry ); + close( wait ); + + for ( i; Tasks ) { + delete(t[i]); + } + + // sout | total_operations; + + for ( i; Processors ) { + delete(proc[i]); + } + sout | "done"; + return 0; +} Index: tests/concurrent/channels/churn.cfa =================================================================== --- tests/concurrent/channels/churn.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/churn.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,145 @@ +#include +#include +#include +#include +#include "channel.hfa" +#include +#include +#include +size_t Processors = 1, Channels = 4, Producers = 2, Consumers = 2, ChannelSize = 128; + +owner_lock o; + +size_t total_operations = 0; +size_t cons_check = 0, prod_check = 0; + +channel( size_t ) * channels; +thread Consumer {}; + +void getRandArray( int * chanIndices ) { + for ( int i = 0; i < Channels; i++ ) { + chanIndices[i] = i; + } + for ( int i = 0; i < Channels; i++ ) { + int loc_1 = prng() % Channels; + int loc_2 = prng() % Channels; + int temp = chanIndices[ loc_1 ]; + chanIndices[ loc_1 ] = chanIndices[ loc_2 ]; + chanIndices[ loc_2 ] = temp; + } +} + +void main(Consumer & this) { + size_t i = 0; + size_t runs = 0; + size_t my_check = 0; + int chanIndices[Channels]; + getRandArray( chanIndices ); + try { + for ( ;;i++ ) { + size_t j = remove( channels[ chanIndices[ i % Channels ] ] ); + my_check = my_check ^ j; + runs++; + } + } catch ( channel_closed * e ) {} + + // flush out rest of channels + for ( i; Channels ) { + try { + for ( ;; ) { + size_t j = remove( channels[ i ] ); + my_check = my_check ^ j; + runs++; + } + } catchResume ( channel_closed * e ) {} // continue to remove until would block + catch ( channel_closed * e ) {} + } + + lock(o); + total_operations += runs; + cons_check = cons_check ^ my_check; + // sout | "Cons: " | runs; + unlock(o); +} + +thread Producer {}; + +void main(Producer & this) { + size_t i = 0; + size_t runs = 0; + size_t my_check = 0; + int chanIndices[Channels]; + getRandArray( chanIndices ); + try { + for ( ;;i++ ) { + insert( channels[ chanIndices[ i % Channels ] ], i ); + my_check = my_check ^ i; + runs++; + } + } catch ( channel_closed * e ) {} + lock(o); + total_operations += runs; + prod_check = prod_check ^ my_check; + // sout | "Prods: " | runs; + unlock(o); +} + + +int main( int argc, char *argv[] ) { + switch( argc ) { + case 4: + if ( strcmp( argv[3], "d" ) != 0 ) { // default ? + if ( atoi( argv[3] ) < 1 ) goto Usage; + ChannelSize = atoi( argv[3] ); + } // if + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + if ( atoi( argv[2] ) < 1 ) goto Usage; + Channels = 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 ]" + | " [ producers > 0 | d ]" + | " [ consumers > 0 | d ]" + | " [ channels > 0 | d ]"; + exit( EXIT_FAILURE ); + } + processor p[Processors - 1]; + channels = aalloc( Channels ); + + // sout | "Processors: " | Processors | " Producers: " | Producers | " Consumers: " | Consumers | "Channels: " | Channels | " Channel Size: " | ChannelSize; + + for ( i; Channels ) + channels[i]{ ChannelSize }; + + sout | "start"; + { + Consumer c[Consumers]; + { + Producer p[Producers]; + sleep(10`s); + for ( i; Channels ) + close( channels[i] ); + } + } + + adelete( channels ); + + if ( cons_check != prod_check ) + sout | "CHECKSUM MISMATCH !!!"; + + // sout | total_operations; + // print_stats_now( *active_cluster(), CFA_STATS_READY_Q ); + + sout | "done"; + return 0; +} Index: tests/concurrent/channels/contend.cfa =================================================================== --- tests/concurrent/channels/contend.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/contend.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,147 @@ +#include +#include +#include +#include +#include +#include +#include + +owner_lock o; + +size_t total_operations = 0; + +typedef channel( size_t ) Channel; + +Channel * channels; + +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) { + size_t runs = 0; + size_t my_check = 0; + try { + for ( ;; ) { + size_t j = remove( channels[ this.i ] ); + my_check = my_check ^ j; + runs++; + } + } catchResume ( channel_closed * e ) {} // continue to remove until would block + catch ( channel_closed * e ) {} + 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) { + size_t runs = 0; + size_t my_check = 0; + try { + for ( ;; ) { + insert( channels[ this.i ], (size_t)runs ); + my_check = my_check ^ ((size_t)runs); + runs++; + } + } catch ( channel_closed * e ) {} + lock(o); + total_operations += runs; + prod_check = prod_check ^ my_check; + // sout | "P: " | runs; + unlock(o); +} + +static inline int test( size_t Processors, size_t Channels, size_t Producers, size_t Consumers, size_t ChannelSize ) { + size_t Clusters = Channels; + // create a cluster + cluster clus[Clusters]; + processor * proc[Processors]; + for ( i; Processors ) { + (*(proc[i] = malloc())){clus[i % Clusters]}; + } + + channels = aalloc( 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 ( j; Channels ) { + for ( i; Producers ) { + (*(p[i] = malloc())){ j, clus[j % Clusters] }; + } + + for ( i; Consumers ) { + (*(c[i] = malloc())){ j, clus[j % Clusters] }; + } + } + + sleep(10`s); + + for ( i; Channels ) + close( channels[i] ); + + for ( i; Producers * Channels ) { + delete(p[i]); + } + for ( i; Consumers * Channels ) { + delete(c[i]); + } + + adelete( channels ); + // sout | 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; +} + +int main( int argc, char * argv[] ) { + size_t Processors = 1, Channels = 1, Producers = 4, Consumers = 4, ChannelSize = 128; + switch ( argc ) { + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + ChannelSize = atoi( argv[2] ); + } // if + case 2: + if ( strcmp( argv[1], "d" ) != 0 ) { // default ? + Processors = atoi( argv[1] ); + if ( Processors < 1 ) goto Usage; + } // if + case 1: // use defaults + break; + default: + Usage: + sout | "Usage: " | argv[0] + | " [ processors (> 0) | 'd' (default " | Processors + | ") ] [ channel size (>= 0) | 'd' (default " | ChannelSize + | ") ]" ; + exit( EXIT_FAILURE ); + } // switch + test(Processors, Channels, Producers, Consumers, ChannelSize); +} Index: tests/concurrent/channels/daisy_chain.cfa =================================================================== --- tests/concurrent/channels/daisy_chain.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/daisy_chain.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,74 @@ +#include +#include +#include +#include +#include +#include +#include + +size_t total_operations = 0; +size_t Processors = 1, Tasks = 4; + +owner_lock o; + +typedef channel( int ) Channel; + +Channel * chain; + +thread Task {}; +void main(Task & this) { + size_t runs = 0; + try{ + for ( ;; ) { + remove( *chain ); + insert( *chain, 0 ); + runs++; + } + } catch ( channel_closed * e ) {} + lock( o ); + total_operations += runs; + unlock( o ); +} + + +int main( int argc, char * argv[] ) { + switch ( argc ) { + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + Tasks = atoi( argv[2] ); + if ( Tasks < 1 ) goto Usage; + } // if + case 2: + if ( strcmp( argv[1], "d" ) != 0 ) { // default ? + Processors = atoi( argv[1] ); + if ( Processors < 1 ) goto Usage; + } // if + case 1: // use defaults + break; + default: + Usage: + sout | "Usage: " | argv[0] + | " [ processors (> 0) | 'd' (default " | Processors + | ") ] [ channel size (>= 0) | 'd' (default " | Tasks + | ") ]" ; + exit( EXIT_FAILURE ); + } // switch + processor proc[Processors - 1]; + + sout | "start"; + Channel chainChan{ 1 }; + + insert( chainChan, 0 ); + + chain = &chainChan; + { + Task t[Tasks]; + sleep(10`s); + close( chainChan ); + } + + // sout | total_operations; + sout | "done"; + + return 0; +} Index: tests/concurrent/channels/hot_potato.cfa =================================================================== --- tests/concurrent/channels/hot_potato.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/hot_potato.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,88 @@ +#include +#include +#include +#include +#include +#include +#include + +size_t total_operations = 0; +size_t Processors = 1, Tasks = 4; + +owner_lock o; + +typedef channel( int ) Channel; + +Channel * chain; + +bool done = false; +int id_counter = 0; +thread Task { int id; int right; }; +static inline void ?{}( Task & this ) with(this) { + id = __atomic_fetch_add( &id_counter, 1, __ATOMIC_SEQ_CST ); + right = (id + 1) % Tasks; +} +void main(Task & this) with(this) { + size_t runs = 0; + try { + for ( ;; ) { + if ( done ) break; + remove( chain[id] ); + insert( chain[right], 0 ); + runs++; + } + } catch ( channel_closed * e ) {} + lock( o ); + total_operations += runs; + unlock( o ); +} + + +int main( int argc, char * argv[] ) { + switch ( argc ) { + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + Tasks = atoi( argv[2] ); + if ( Tasks < 1 ) goto Usage; + } // if + case 2: + if ( strcmp( argv[1], "d" ) != 0 ) { // default ? + Processors = atoi( argv[1] ); + if ( Processors < 1 ) goto Usage; + } // if + case 1: // use defaults + break; + default: + Usage: + sout | "Usage: " | argv[0] + | " [ processors (> 0) | 'd' (default " | Processors + | ") ] [ channel size (>= 0) | 'd' (default " | Tasks + | ") ]" ; + exit( EXIT_FAILURE ); + } // switch + processor proc[Processors - 1]; + + sout | "start"; + + chain = aalloc( Tasks ); + for ( i; Tasks ) { + chain[i]{ 1 }; + } + + insert( chain[0], 0 ); + { + Task t[Tasks]; + sleep(10`s); + done = true; + for ( j; Tasks ) + close( chain[j] ); + } + + // sout | total_operations; + + sout | "done"; + + adelete(chain); + + return 0; +} Index: tests/concurrent/channels/ping_pong.cfa =================================================================== --- tests/concurrent/channels/ping_pong.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/ping_pong.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,64 @@ +#include +#include +#include +#include +#include +#include +#include + +typedef channel( int ) Channel; + +Channel * ping; +Channel * pong; + +bool done = false; +size_t total_operations = 0; + +thread Pong {}; +void main(Pong & this) { + try { + for ( ;; ) { + if ( done ) break; + insert( *ping, 0 ); + remove( *pong ); + } + } catch ( channel_closed * e ) {} +} + +thread Ping {}; +void main(Ping & this) { + try { + for ( ;; ) { + if ( done ) break; + remove( *ping ); + insert( *pong, 1 ); + total_operations++; + } + } catch ( channel_closed * e ) {} +} + + +int main( int argc, char * argv[] ) { + sout | "start"; + processor proc[1]; + + Channel pingChan{ 1 }; + Channel pongChan{ 1 }; + + ping = &pingChan; + pong = &pongChan; + + { + Ping pi; + Pong po; + sleep(10`s); + done = true; + close( *pong ); + close( *ping ); + } + + // sout | total_operations; + + sout | "done"; + return 0; +} Index: tests/concurrent/channels/pub_sub.cfa =================================================================== --- tests/concurrent/channels/pub_sub.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) +++ tests/concurrent/channels/pub_sub.cfa (revision 9082d7e82d9a7493aed9e548946de1837caa325a) @@ -0,0 +1,155 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +size_t total_operations = 0; +size_t Processors = 1, Tasks = 4; + +typedef channel( size_t ) Channel; + +channel( int ) * barWait; +channel( int ) * entryWait; +int BarrierSize = 4; +static inline void closeBarrier() { + close(*entryWait); + close(*barWait); +} + +static inline void initBarrier() { + barWait = malloc(); + entryWait = malloc(); + (*barWait){ BarrierSize }; + (*entryWait){ BarrierSize }; + for ( j; BarrierSize ) + insert( *entryWait, j ); +} + +static inline void deleteBarrier() { + delete(barWait); + delete(entryWait); +} + +static inline void barrier() { + int ticket = remove( *entryWait ); + if ( ticket == BarrierSize - 1 ) { + for ( j; BarrierSize - 1 ) + insert( *barWait, j ); + return; + } + ticket = remove( *barWait ); + + // last one out + if ( BarrierSize == 1 || ticket == BarrierSize - 2 ) { + for ( j; BarrierSize ) + insert( *entryWait, j ); + } +} + +Channel * chans; +owner_lock o; + +thread Task { size_t id; }; +static inline void ?{}( Task & p, size_t i, cluster & clu ) { + ((thread &)p){ clu }; + p.id = i; +} +void main(Task & this) with(this) { + size_t runs = 0; + try { + for ( ;; ) { + // publish + for ( i; Tasks ) { + insert(chans[id], i); + } + + // subscribe + for ( i; Tasks ) { + remove( chans[i] ); + } + barrier(); + runs++; + } + } catch ( channel_closed * e ) { } + lock(o); + total_operations += runs; + // sout | runs; + unlock(o); +} + + +int main( int argc, char * argv[] ) { + switch ( argc ) { + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + Tasks = atoi( argv[2] ); + if ( Tasks < 1 ) goto Usage; + } // if + case 2: + if ( strcmp( argv[1], "d" ) != 0 ) { // default ? + Processors = atoi( argv[1] ); + if ( Processors < 1 ) goto Usage; + } // if + case 1: // use defaults + break; + default: + Usage: + sout | "Usage: " | argv[0] + | " [ processors (> 0) | 'd' (default " | Processors + | ") ] [ Tasks (> 0) | 'd' (default " | Tasks + | ") ]" ; + exit( EXIT_FAILURE ); + } // switch + BarrierSize = Tasks; + + size_t Clusters = 1; + // create a cluster + cluster clus[Clusters]; + processor * proc[Processors]; + + // setup processors + for ( i; Processors ) + (*(proc[i] = malloc())){clus[i % Clusters]}; + + // setup pub/sub chans + chans = aalloc( Tasks ); + for ( i; Tasks ) + chans[i]{ Tasks }; + + // setup barrier + initBarrier(); + + // sout | "Processors: " | Processors | " ProdsPerChan: " | Producers | " ConsPerChan: " | Consumers | "Channels: " | Channels | " Channel Size: " | ChannelSize; + + sout | "start"; + Task * t[Tasks]; + + // create tasks + for ( i; Tasks ) + (*(t[i] = malloc())){ i, clus[i % Clusters] }; + + sleep(10`s); + + closeBarrier(); + for ( i; Tasks ) + close( chans[i] ); + + for ( i; Tasks ) { + delete(t[i]); + } + + deleteBarrier(); + + // sout | total_operations; + + for ( i; Processors ) { + delete(proc[i]); + } + adelete( chans ); + sout | "done"; + return 0; +}