source: tests/concurrent/channels/parallel_harness.hfa @ 18ea270

ADTast-experimental
Last change on this file since 18ea270 was de934c7, checked in by caparson <caparson@…>, 20 months ago

minor changes to channel test harness

  • Property mode set to 100644
File size: 4.4 KB
Line 
1#include <locks.hfa>
2#include <fstream.hfa>
3#include <stdio.h>
4#include <channel.hfa>
5#include <thread.hfa>
6#include <time.hfa>
7
8// user defines this
9// #define BIG 1
10
11owner_lock o;
12
13unsigned long long total_operations = 0;
14
15struct bigObject {
16    size_t a;
17    size_t b;
18    size_t c;
19    size_t d;
20    size_t e;
21    size_t f;
22    size_t g;
23    size_t h;
24};
25
26void ?{}( bigObject & this, size_t i ) with(this) { a = i; b = i; c = i; d = i; e = i; f = i; g = i; h = i; }
27void ?{}( bigObject & this ) { this{0}; }
28
29#ifdef BIG
30typedef channel( bigObject ) Channel;
31#else
32typedef channel( size_t ) Channel;
33#endif
34
35Channel * channels;
36
37volatile bool cons_done = false, prod_done = false;
38volatile int cons_done_count = 0;
39size_t cons_check = 0, prod_check = 0;
40
41thread Consumer {
42    size_t i;
43};
44static inline void ?{}( Consumer & c, size_t i, cluster & clu ) {
45    ((thread &)c){ clu };
46    c.i = i;
47}
48void main(Consumer & this) {
49    unsigned long long runs = 0;
50    size_t my_check = 0;
51    for ( ;; ) {
52        if ( cons_done ) break;
53        #ifdef BIG
54        bigObject j = remove( channels[ this.i ] );
55        my_check = my_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
56        #else
57        size_t j = remove( channels[ this.i ] );
58        my_check = my_check ^ j;
59        #endif
60       
61        if ( !prod_done ) runs++;
62    }
63    lock(o);
64    total_operations += runs;
65    cons_done_count++;
66    cons_check = cons_check ^ my_check;
67    // sout | "C: " | runs;
68    unlock(o);
69}
70
71thread Producer {
72    size_t i;
73};
74static inline void ?{}( Producer & p, size_t i, cluster & clu ) {
75    ((thread &)p){ clu };
76    p.i = i;
77}
78void main(Producer & this) {
79    unsigned long long runs = 0;
80    size_t my_check = 0;
81    for ( ;; ) {
82        if ( prod_done ) break;
83        #ifdef BIG
84        bigObject j{(size_t)runs};
85        insert( channels[ this.i ], j );
86        my_check = my_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
87        #else
88        insert( channels[ this.i ], (size_t)runs );
89        my_check = my_check ^ ((size_t)runs);
90        #endif
91        runs++;
92    }
93    lock(o);
94    total_operations += runs;
95    prod_check = prod_check ^ my_check;
96    // sout | "P: " | runs;
97    unlock(o);
98}
99
100
101int test( size_t Processors, size_t Channels, size_t Producers, size_t Consumers, size_t ChannelSize ) {
102    size_t Clusters = Processors;
103    // create a cluster
104    cluster clus[Clusters];
105    processor * proc[Processors];
106    for ( i; Processors ) {
107        (*(proc[i] = alloc())){clus[i % Clusters]};
108    }
109
110    channels = aalloc( Channels );
111
112    // sout | "Processors: " | Processors | " ProdsPerChan: " | Producers | " ConsPerChan: " | Consumers | "Channels: " | Channels | " Channel Size: " | ChannelSize;
113   
114    for ( i; Channels ) {
115        channels[i]{ ChannelSize };
116    }
117
118    sout | "start";
119    Consumer * c[Consumers * Channels];
120    Producer * p[Producers * Channels];
121
122    for ( i; Consumers * Channels ) {
123        (*(c[i] = alloc())){ i % Channels, clus[i % Clusters] };
124    }
125
126    for ( i; Producers * Channels ) {
127        (*(p[i] = alloc())){ i % Channels, clus[i % Clusters] };
128    }
129
130    sleep(10`s);
131    prod_done = true;
132
133    for ( i; Producers * Channels ) {
134        delete(p[i]);
135    }
136
137    sout | "prods";
138    cons_done = true;
139    while( cons_done_count != Consumers * Channels ) {
140        for ( i; Channels ) {
141            if ( has_waiters( channels[i] ) ){
142                #ifdef BIG
143                bigObject b{0};
144                insert( channels[i], b );
145                #else
146                insert( channels[i], 0 );
147                #endif
148            }
149        }
150       
151    }
152
153    sout | "cons";
154    for ( i; Consumers * Channels ) {
155        delete(c[i]);
156    }
157
158    sout | "flush";
159    for ( i; Channels ) {
160        for ( ;; ) {
161            if ( get_count( channels[i] ) > 0 ) {
162                #ifdef BIG
163                bigObject j = remove( channels[ i ] );
164                cons_check = cons_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
165                #else
166                size_t j = remove( channels[ i ] );
167                cons_check = cons_check ^ j;
168                #endif
169            } else break;
170        }
171    }
172
173    adelete( channels );
174    // sout | "total channel ops: " | total_operations;
175    if ( cons_check != prod_check )
176        sout | "CHECKSUM MISMATCH !!!";
177    // print_stats_now( *active_cluster(), CFA_STATS_READY_Q);
178
179    for ( i; Processors ) {
180        delete(proc[i]);
181    }
182    sout | "done";
183    return 0;
184}
Note: See TracBrowser for help on using the repository browser.