1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
|
#ifndef _plan9_
#include <signal.h>
#include <stdio.h>
#endif
#include <xthread.h>
#ifndef __BEOS__
struct taskinfo definfo = {
0,
};
#ifndef nthreads
#define getrange1(range,ncpu) ((range)*(ncpu))/nthreads
#define getrange2(range,ncpu) getrange1(range,(ncpu)+1)
int ethreads = 0;
int nthreads = 1;
#ifdef USE_PTHREAD
/* Well conde follows is probably very ugly, since this is
* my absolutely first application for threads. Please let
* me know how to improvie it
*/
static pthread_cond_t synccond, startcond;
static struct taskinfo infos[MAXTHREADS];
static pthread_mutex_t synccondmutex, startcondmutex;
pthread_mutex_t semaphors[MAXSEMAPHORS];
pthread_cond_t conds[MAXCONDS];
/*This loop is executed whole time in slave threads.
Its function is following:
1) wait for message
2) call function from message
3) syncronize
4) again
To invoke this mechanizm main thread(#1) should call
xth_function
xth_synchronize forces forces thread to wait for others
*/
static int nfinished;
static int npending;
static int counter;
static int bcounter;
static int bcounter1;
static int range;
static void *data;
static xfunction function;
static void *control_routine(void *i)
{
struct taskinfo *info = i;
int mycounter = 0;
int r;
void *d;
xfunction f;
while (1) {
/* quite a lot pthread calls. Please if you are
* reading this code and crying "OH NO!" so ugly
* handling! Why so much calls? Stop crying, I am
* newbie in threads. Please rewrite my code and
* send me better and faster version.
*
* This function comunicates with pth_function from main loop
* as follows: it uses startcond to wait for order start function!
* Counter is used to ensure that main function did not give
* order whie control_routine was busy
*
* after order is received, function adress is readed from global
* variables and started. Pth_function then executes its
* own part of calculation. After that it waits counter
* nfinished to reach number of thasks-1. This is done
* using cond synccond and synccondmutex. Quite complex
* but it seems to work. Looking forward for someone, who
* should reduce number of _lock/_unlock
*/
pthread_mutex_lock(&synccondmutex);
nfinished++;
pthread_cond_signal(&synccond);
pthread_mutex_unlock(&synccondmutex);
pthread_mutex_lock(&startcondmutex);
while (mycounter >= counter) {
if (bcounter > bcounter1) {
/*Well we are already locked using start lock..should be OK */
mycounter--;
bcounter1++;
break;
}
pthread_cond_wait(&startcond, &startcondmutex);
}
r = range;
d = data;
f = function;
npending--;
pthread_mutex_unlock(&startcondmutex);
mycounter++;
f(d, info, getrange1(r, info->n), getrange2(r, info->n));
}
return NULL;
}
void pth_init(int nthreads1)
{
int i;
pthread_attr_t attr;
if (ethreads)
return;
if (nthreads1 == 1 || nthreads1 == 0)
return; /*use nothreads_* calls */
if (nthreads1 > MAXTHREADS)
nthreads1 = MAXTHREADS;
nthreads = nthreads1;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
if (pthread_cond_init(&synccond, NULL))
exit(1);
if (pthread_cond_init(&startcond, NULL))
exit(1);
if (pthread_mutex_init(&synccondmutex, NULL))
exit(1);
if (pthread_mutex_init(&startcondmutex, NULL))
exit(1);
infos[0].n = 0;
/*infos[0].id = pthread_self(); */
for (i = 0; i < MAXSEMAPHORS; i++) {
pthread_mutex_init(semaphors + i, NULL);
}
for (i = 0; i < MAXCONDS; i++) {
pthread_cond_init(conds + i, NULL);
}
nfinished = 0;
for (i = 0; i < nthreads - 1; i++) {
if (pthread_create
(&infos[i + 1].id, &attr, control_routine, infos + i + 1)) {
nthreads = i + 1;
break;
}
infos[i + 1].n = i + 1;
}
if (nthreads != 1)
ethreads = 1;
}
void pth_synchronize()
{
/*Our job is done, synchronize now */
if (nfinished < nthreads - 1) {
pthread_mutex_lock(&synccondmutex);
while (nfinished < nthreads - 1) {
pthread_cond_wait(&synccond, &synccondmutex);
}
pthread_mutex_unlock(&synccondmutex);
}
/*Ok job is done, lets continue :) */
}
void pth_bgjob(xfunction f, void *d)
{
pthread_mutex_lock(&startcondmutex);
if (npending) {
printf("Collision!\n"); /*FIXME:remove this..I just want to know how often this happends */
pthread_mutex_unlock(&startcondmutex);
f(d, infos, 0, 0);
}
if (bcounter < bcounter1) {
printf("Internal error\a\n");
}
if (!nfinished) {
pthread_mutex_unlock(&startcondmutex);
/*no more CPU available :( */
f(d, infos, 0, 0);
}
data = d;
range = 0;
function = f;
bcounter++;
nfinished--;
npending++;
pthread_cond_signal(&startcond);
pthread_mutex_unlock(&startcondmutex);
}
void pth_function(xfunction f, void *d, int r)
{
pth_synchronize();
pthread_mutex_lock(&startcondmutex);
data = d;
range = r;
function = f;
/*And lets start it:) */
nfinished = 0;
npending = nthreads - 1;
counter++;
if (nthreads == 2)
pthread_cond_signal(&startcond);
else
pthread_cond_broadcast(&startcond);
pthread_mutex_unlock(&startcondmutex);
function(data, infos, getrange1(range, 0), getrange2(range, 0));
}
void pth_uninit()
{
/*Should be empty for now since all threads will be killed after exit call */
/*FIXME should be added something if necessary :) */
nthreads = 1;
ethreads = 0;
}
#endif /*POSIX threads */
#endif /*nthreads */
#endif /*__BEOS__*/
|
