Index: tests/concurrency/futures/.expect/abandon.txt =================================================================== --- tests/concurrency/futures/.expect/abandon.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/.expect/abandon.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrency/futures/.expect/basic.txt =================================================================== --- tests/concurrency/futures/.expect/basic.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/.expect/basic.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrency/futures/.expect/multi.txt =================================================================== --- tests/concurrency/futures/.expect/multi.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/.expect/multi.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrency/futures/.expect/select_future.txt =================================================================== --- tests/concurrency/futures/.expect/select_future.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/.expect/select_future.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,6 @@ +start 1: blocking path future test +done 1 +start 2: nonblocking path future test +done 2 +start 3: try_get future test +done 3 Index: tests/concurrency/futures/.expect/typed.txt =================================================================== --- tests/concurrency/futures/.expect/typed.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/.expect/typed.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrency/futures/.expect/wait_any.txt =================================================================== --- tests/concurrency/futures/.expect/wait_any.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/.expect/wait_any.txt (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,2 @@ +start +done Index: tests/concurrency/futures/abandon.cfa =================================================================== --- tests/concurrency/futures/abandon.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/abandon.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,102 @@ +#include + +enum {NFUTURES = 10}; + +thread Server { + int cnt; + future_t * requests[NFUTURES]; +}; + +void ?{}( Server & this ) { + this.cnt = 0; + for(i; NFUTURES) { + this.requests[i] = 0p; + } +} + +void ^?{}( Server & mutex this ) { + assert(this.cnt == 0); + for(i; NFUTURES) { + this.requests[i] = 0p; + } +} + +void process( Server & this, int i ) { + if( this.requests[i] == 0p ) return; + future_t * f = this.requests[i]; + this.requests[i] = 0p; + this.cnt--; + fulfil( *f ); +} + +void call( Server & mutex this, future_t & f ) { + for(i; NFUTURES) { + if( this.requests[i] == 0p ) { + this.requests[i] = &f; + this.cnt++; + return; + } + } + abort("Monitor Error"); +} + +void main( Server & this ) { + unsigned i = 0; + for() { + waitfor( ^?{} : this ) { + break; + } + or when( this.cnt < NFUTURES ) waitfor( call: this ) {} + or else { + process( this, i % NFUTURES ); + i++; + } + } + + for(i; NFUTURES) { + process( this, i ); + } +} + +Server * the_server; +thread Worker {}; + +void thrash(void) { + volatile int locals[250]; + for(i; 250) { + locals[i] = 0xdeadbeef; + } +} + +void work(int i) { + future_t *mine = new(); + call( *the_server, *mine ); + if (i % 2 == 0) { + abandon( *mine ); + } else { + wait( *mine ); + delete( mine ); + } +} + +void main( Worker & ) { + for(i;150) { + thrash(); + work(i); + thrash(); + } +} + +int main() { + printf( "start\n" ); // non-empty .expect file + processor procs[2]; + { + Server server; + the_server = &server; + { + Worker workers[17]; + } + } + printf( "done\n" ); // non-empty .expect file + +} Index: tests/concurrency/futures/basic.cfa =================================================================== --- tests/concurrency/futures/basic.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/basic.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,97 @@ +#include + +enum {NFUTURES = 10}; + +thread Server { + int cnt; + future_t * requests[NFUTURES]; +}; + +void ?{}( Server & this ) { + this.cnt = 0; + for(i; NFUTURES) { + this.requests[i] = 0p; + } +} + +void ^?{}( Server & mutex this ) { + assert(this.cnt == 0); + for(i; NFUTURES) { + this.requests[i] = 0p; + } +} + +void process( Server & this, int i ) { + if( this.requests[i] == 0p ) return; + future_t * f = this.requests[i]; + this.requests[i] = 0p; + this.cnt--; + fulfil( *f ); +} + +void call( Server & mutex this, future_t & f ) { + for(i; NFUTURES) { + if( this.requests[i] == 0p ) { + this.requests[i] = &f; + this.cnt++; + return; + } + } + abort("Monitor Error"); +} + +void main( Server & this ) { + unsigned i = 0; + for() { + waitfor( ^?{} : this ) { + break; + } + or when( this.cnt < NFUTURES ) waitfor( call: this ) {} + or else { + process( this, i % NFUTURES ); + i++; + } + } + + for(i; NFUTURES) { + process( this, i ); + } +} + +Server * the_server; +thread Worker {}; + +void thrash(void) { + volatile int locals[250]; + for(i; 250) { + locals[i] = 0xdeadbeef; + } +} + +void work(void) { + future_t mine; + call( *the_server, mine ); + wait( mine ); +} + +void main( Worker & ) { + for(150) { + thrash(); + work(); + thrash(); + } +} + +int main() { + printf( "start\n" ); // non-empty .expect file + processor procs[2]; + { + Server server; + the_server = &server; + { + Worker workers[17]; + } + } + printf( "done\n" ); // non-empty .expect file + +} Index: tests/concurrency/futures/multi.cfa =================================================================== --- tests/concurrency/futures/multi.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/multi.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,104 @@ +#include +#include + +enum {NFUTURES = 10}; + +thread Server { + int pending, done, iteration; + multi_future(int) * request; +}; + +void ?{}( Server & this ) { + ((thread&)this){"Server Thread"}; + this.pending = 0; + this.done = 0; + this.iteration = 0; + this.request = 0p; +} + +void ^?{}( Server & mutex this ) { + assert(this.pending == 0); + this.request = 0p; +} + +void init( Server & this , multi_future(int) * f ) { + this.request = f; +} + +void call( Server & mutex this ) { + this.pending++; +} + +void finish( Server & mutex this ) { + this.done++; +} + +void main( Server & this ) { + MAIN_LOOP: + for() { + waitfor( ^?{} : this ) { + break; + } + or waitfor( call: this ) { + if (this.pending != NFUTURES) { continue MAIN_LOOP; } + + this.pending = 0; + fulfil( *this.request, this.iteration ); + this.iteration++; + + for(NFUTURES) { + waitfor( finish: this ); + } + + reset( *this.request ); + this.done = 0; + } + } + +} + +Server * the_server; +thread Worker {}; +void ?{}(Worker & this) { + ((thread&)this){"Worker Thread"}; +} + +multi_future(int) * shared_future; + +void thrash(void) { + volatile int locals[250]; + for(i; 250) { + locals[i] = 0xdeadbeef; + } +} + +void work(int num) { + call( *the_server ); + int res = wait( *shared_future ); + if( res != num ) abort(); + finish( *the_server ); +} + +void main( Worker & ) { + for (i; 10) { + thrash(); + work(i); + thrash(); + } +} + +int main() { + printf( "start\n" ); // non-empty .expect file + processor procs[2]; + shared_future = new(); + { + Server server; + the_server = &server; + init(server, shared_future); + { + Worker workers[NFUTURES]; + } + } + delete( shared_future ); + printf( "done\n" ); // non-empty .expect file +} Index: tests/concurrency/futures/select_future.cfa =================================================================== --- tests/concurrency/futures/select_future.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/select_future.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,198 @@ +#include +#include +#include + +enum {NFUTURES = 10}; + +thread Server { + int pending, done, iteration; + future(int) * request; +}; + +void ?{}( Server & this ) { + ((thread&)this){"Server Thread"}; + this.pending = 0; + this.done = 0; + this.iteration = 0; + this.request = 0p; +} + +void ^?{}( Server & mutex this ) { + assert(this.pending == 0); + this.request = 0p; +} + +void init( Server & this , future(int) * f ) { + this.request = f; +} + +void call( Server & mutex this ) { + this.pending++; +} + +void finish( Server & mutex this ) { + this.done++; +} + +void main( Server & this ) { + MAIN_LOOP: + for() { + waitfor( ^?{} : this ) { + break; + } + or waitfor( call: this ) { + if (this.pending != NFUTURES) { continue MAIN_LOOP; } + + this.pending = 0; + fulfil( *this.request, this.iteration ); + this.iteration++; + + for(NFUTURES) { + waitfor( finish: this ); + } + + reset( *this.request ); + this.done = 0; + } + } + +} + +Server * the_server; +thread Worker {}; +void ?{}(Worker & this) { + ((thread&)this){"Worker Thread"}; +} + +future(int) * shared_future; + +void thrash(void) { + volatile int locals[250]; + for(i; 250) { + locals[i] = 0xdeadbeef; + } +} + +void work(int num) { + call( *the_server ); + int res = get( *shared_future ); + if( res != num ) abort(); + finish( *the_server ); +} + +void main( Worker & ) { + for (i; 10) { + thrash(); + work(i); + thrash(); + } +} + +thread Worker2 {}; + +semaphore before{0}; +semaphore after_server{0}; +semaphore after_worker{0}; + +void work2( int num ) { + P( before ); + int res = get( *shared_future ); + if( res != num ) abort(); + V( after_server ); + P( after_worker ); +} + +void main( Worker2 & ) { + for (i; 10) { + thrash(); + work2(i); + thrash(); + } +} + +thread Server2 {}; + +void main( Server2 & ) { + for (i; 10) { + fulfil( *shared_future , i ); + V( before, NFUTURES ); + for ( i; NFUTURES ) P( after_server ); + reset( *shared_future ); + V( after_worker, NFUTURES ); + } +} + +barrier bar = { NFUTURES + 1 }; + +thread Worker3 {}; + +void work3( int num ) { + [int, bool] tt; + do { + tt = try_get( *shared_future ); + } while ( ! tt.1 ); + if( tt.0 != num ) abort(); + V( after_server ); + block(bar); +} + +void main( Worker3 & ) { + for (i; 10) { + thrash(); + work3(i); + thrash(); + } +} + +thread Server3 {}; + +void main( Server3 & ) { + for (i; 10) { + fulfil( *shared_future , i ); + for ( i; NFUTURES ) P( after_server ); + reset( *shared_future ); + block(bar); + } +} + +int main() { + printf( "start 1: blocking path future test\n" ); + processor procs[11]; + shared_future = new(); + { + Server server; + the_server = &server; + init(server, shared_future); + { + Worker workers[NFUTURES]; + } + } + delete( shared_future ); + printf( "done 1\n" ); + + printf( "start 2: nonblocking path future test\n" ); + shared_future = new(); + + { + Server2 server; + { + Worker2 workers[NFUTURES]; + } + } + + delete( shared_future ); + printf( "done 2\n" ); + + printf( "start 3: try_get future test\n" ); + shared_future = new(); + + { + Worker3 workers[NFUTURES]; + { + Server3 server; + } + } + + delete( shared_future ); + printf( "done 3\n" ); +} Index: tests/concurrency/futures/typed.cfa =================================================================== --- tests/concurrency/futures/typed.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/typed.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,97 @@ +#include +#include +enum {NFUTURES = 10}; + +thread Server { + int cnt; + single_future(int) * requests[NFUTURES]; +}; + +void ?{}( Server & this ) { + this.cnt = 0; + for(i; NFUTURES) { + this.requests[i] = 0p; + } +} + +void ^?{}( Server & mutex this ) { + assert(this.cnt == 0); + for(i; NFUTURES) { + this.requests[i] = 0p; + } +} + +void process( Server & this, int i ) { + if( this.requests[i] == 0p ) return; + single_future(int) * f = this.requests[i]; + this.requests[i] = 0p; + this.cnt--; + fulfil( *f , i); +} + +void call( Server & mutex this, single_future(int) & f ) { + for(i; NFUTURES) { + if( this.requests[i] == 0p ) { + this.requests[i] = &f; + this.cnt++; + return; + } + } + abort("Monitor Error"); +} + +void main( Server & this ) { + unsigned i = 0; + for() { + waitfor( ^?{} : this ) { + break; + } + or when( this.cnt < NFUTURES ) waitfor( call: this ) {} + or else { + process( this, i % NFUTURES ); + i++; + } + } + + for(i; NFUTURES) { + process( this, i ); + } +} + +Server * the_server; +thread Worker {}; + +void thrash(void) { + volatile int locals[250]; + for(i; 250) { + locals[i] = 0xdeadbeef; + } +} + +void work(void) { + single_future(int) mine; + call( *the_server, mine ); + wait( mine ); +} + +void main( Worker & ) { + for(150) { + thrash(); + work(); + thrash(); + } +} + +int main() { + printf( "start\n" ); // non-empty .expect file + processor procs[2]; + { + Server server; + the_server = &server; + { + Worker workers[17]; + } + } + printf( "done\n" ); // non-empty .expect file + +} Index: tests/concurrency/futures/wait_any.cfa =================================================================== --- tests/concurrency/futures/wait_any.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) +++ tests/concurrency/futures/wait_any.cfa (revision c26bea2aa0f87b4b070349c5801adc32fb0d4cf9) @@ -0,0 +1,77 @@ +#include +#include +#include +#include +#include + +simple_owner_lock l; +pthread_cond_var( simple_owner_lock ) c; + +size_t num_futures = 10; + +future_t * futures; + +size_t * shuffle_arr; + +size_t numtimes = 10; + +volatile bool done = false; + +void synchronize() { + lock(l); + if (empty(c)) { + wait(c,l); + } else { + for ( i; num_futures ) { + reset(futures[i]); + } + notify_one(c); + } + unlock(l); +} + +thread Waiter {}; +void main( Waiter & this ) { + for (numtimes) { + wait_any(futures, num_futures); + synchronize(); + } + done = true; + while (done) notify_one(c); +} + +thread Deliverer {}; +void main( Deliverer & this ) { + while (!done) { + size_t num_satisfy = random(1,num_futures); + for ( i; num_satisfy ) { // random shuffle a few values + swap( shuffle_arr[random(num_futures)], shuffle_arr[random(num_futures)] ); + } + for ( i; num_satisfy ) { + fulfil(futures[shuffle_arr[i]]); + } + synchronize(); + } + done = false; +} + +int main() { + sout | "start"; + futures = alloc( num_futures ); + shuffle_arr = alloc( num_futures ); + for ( i; num_futures ) { // random shuffle a few values + futures[i]{}; + swap( shuffle_arr[random(num_futures)], shuffle_arr[random(num_futures)] ); + } + for ( i; num_futures ) { + shuffle_arr[i] = i; + } + processor procs[1]; + { + Waiter w; + Deliverer d; + } + free( futures ); + free( shuffle_arr ); + sout | "done"; +}