source: tests/concurrent/channels/parallel_harness.hfa @ b2f3880

ADTast-experimental
Last change on this file since b2f3880 was b2f3880, checked in by Peter A. Buhr <pabuhr@…>, 19 months ago

add newline at end of file

  • Property mode set to 100644
File size: 4.9 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 = 1;
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 = anew( 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    // for ( i; Channels ) {
153    //     // sout | get_count( channels[i] );
154    //     if ( get_count( channels[i] ) < Consumers ){
155    //         #ifdef BIG
156    //         bigObject b{0};
157    //         #endif
158    //         for ( j; Consumers ) {
159    //             #ifdef BIG
160    //             insert( channels[i], b );
161    //             #else
162    //             insert( channels[i], 0 );
163    //             #endif
164    //         }
165    //     }
166    // }
167    sout | "cons";
168    for ( i; Consumers * Channels ) {
169        delete(c[i]);
170    }
171
172    sout | "flush";
173    for ( i; Channels ) {
174        for ( ;; ) {
175            if ( get_count( channels[i] ) > 0 ) {
176                #ifdef BIG
177                bigObject j = remove( channels[ i ] );
178                cons_check = cons_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
179                #else
180                size_t j = remove( channels[ i ] );
181                cons_check = cons_check ^ j;
182                #endif
183            } else break;
184        }
185    }
186
187    adelete( channels );
188    // sout | "total channel ops: " | total_operations;
189    if ( cons_check != prod_check )
190        sout | "CHECKSUM MISMATCH !!!";
191    // print_stats_now( *active_cluster(), CFA_STATS_READY_Q);
192
193    for ( i; Processors ) {
194        delete(proc[i]);
195    }
196    sout | "done";
197    return 0;
198}
Note: See TracBrowser for help on using the repository browser.