Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Make those functions static
[simgrid.git] / src / xbt / parmap.c
1 /* Copyright (c) 2004, 2005, 2007, 2009, 2010. The SimGrid Team.
2  * All rights reserved.                                                     */
3
4 /* This program is free software; you can redistribute it and/or modify it
5  * under the terms of the license (GNU LGPL) which comes with this package. */
6 #include "gras_config.h"
7 #include <unistd.h>
8
9 #ifndef _XBT_WIN32
10 #include <sys/syscall.h>
11 #endif
12
13 #ifdef HAVE_FUTEX_H
14 #include <linux/futex.h>
15 #endif
16
17 #include "xbt/parmap.h"
18 #include "xbt/log.h"
19 #include "xbt/function_types.h"
20 #include "xbt/dynar.h"
21 #include "xbt/xbt_os_thread.h"
22 #include "xbt/sysdep.h"
23
24 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
25 XBT_LOG_NEW_SUBCATEGORY(xbt_parmap_unit, xbt_parmap, "parmap unit testing");
26
27 typedef enum {
28   PARMAP_WORK = 0,
29   PARMAP_DESTROY
30 } e_xbt_parmap_flag_t;
31
32 #ifdef HAVE_FUTEX_H
33 typedef struct s_xbt_event {
34   int work;
35   int done;
36   unsigned int thread_counter;
37   unsigned int threads_to_wait;
38 } s_xbt_event_t, *xbt_event_t;
39
40 static void xbt_event_init(xbt_event_t event);
41 static void xbt_event_signal(xbt_event_t event);
42 static void xbt_event_wait(xbt_event_t event);
43 static void xbt_event_end(xbt_event_t event);
44 #endif
45
46 typedef struct s_xbt_parmap {
47   e_xbt_parmap_flag_t status;
48 #ifdef HAVE_FUTEX_H
49   xbt_event_t sync_event;
50 #endif
51   unsigned int num_workers;
52   unsigned int workers_max_id;
53   void_f_pvoid_t fun;
54   xbt_dynar_t data;
55   unsigned int index;
56 } s_xbt_parmap_t;
57
58 static void *_xbt_parmap_worker_main(void *parmap);
59 #ifdef HAVE_FUTEX_H
60 static void futex_wait(int *uaddr, int val);
61 static void futex_wake(int *uaddr, int val);
62 #endif
63
64 xbt_parmap_t xbt_parmap_new(unsigned int num_workers)
65 {
66   unsigned int i;
67   xbt_os_thread_t worker = NULL;
68
69   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
70
71   /* Initialize the thread pool data structure */
72   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
73 #ifdef HAVE_FUTEX_H
74   parmap->sync_event = xbt_new0(s_xbt_event_t, 1);
75 #endif
76   parmap->num_workers = num_workers;
77   parmap->status = PARMAP_WORK;
78 #ifdef HAVE_FUTEX_H
79   parmap->sync_event->threads_to_wait = num_workers;
80 #endif
81   /* Create the pool of worker threads */
82   for (i = 0; i < num_workers; i++) {
83     worker = xbt_os_thread_create(NULL, _xbt_parmap_worker_main, parmap, NULL);
84     xbt_os_thread_detach(worker);
85   }
86 #ifdef HAVE_FUTEX_H
87   xbt_event_init(parmap->sync_event);
88 #endif
89   return parmap;
90 }
91
92 void xbt_parmap_destroy(xbt_parmap_t parmap)
93 {
94   parmap->status = PARMAP_DESTROY;
95 #ifdef HAVE_FUTEX_H
96   xbt_event_signal(parmap->sync_event);
97   xbt_free(parmap->sync_event);
98 #endif
99   xbt_free(parmap);
100 }
101
102 void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
103 {
104   /* Assign resources to worker threads*/
105   parmap->fun = fun;
106   parmap->data = data;
107   parmap->index = 0;
108 #ifdef HAVE_FUTEX_H
109   xbt_event_signal(parmap->sync_event);
110 #endif
111   XBT_DEBUG("Job done");
112 }
113
114 void* xbt_parmap_next(xbt_parmap_t parmap)
115 {
116   unsigned int index = __sync_fetch_and_add(&parmap->index, 1);
117   if (index < xbt_dynar_length(parmap->data)) {
118     return xbt_dynar_get_as(parmap->data, index, void*);
119   }
120   return NULL;
121 }
122
123 unsigned long xbt_parmap_get_worker_id(xbt_parmap_t parmap)
124 {
125   return (unsigned long) xbt_os_thread_get_extra_data();
126 }
127
128 static void *_xbt_parmap_worker_main(void *arg)
129 {
130   unsigned int worker_id;
131   xbt_parmap_t parmap = (xbt_parmap_t) arg;
132
133   /* Fetch a worker id */
134   worker_id = __sync_fetch_and_add(&parmap->workers_max_id, 1);
135   xbt_os_thread_set_extra_data((void*) (unsigned long) worker_id);
136
137   XBT_DEBUG("New worker thread created (%u)", worker_id);
138   
139   /* Worker's main loop */
140   while (1) {
141 #ifdef HAVE_FUTEX_H
142     xbt_event_wait(parmap->sync_event);
143 #endif
144     if (parmap->status == PARMAP_WORK) {
145
146       XBT_DEBUG("Worker %u got a job", worker_id);
147
148       void* work = xbt_parmap_next(parmap);
149       if (work != NULL) {
150         parmap->fun(work);
151       }
152
153       XBT_DEBUG("Worker %u has finished", worker_id);
154
155     /* We are destroying the parmap */
156     } else {
157 #ifdef HAVE_FUTEX_H
158       xbt_event_end(parmap->sync_event);
159 #endif
160       XBT_DEBUG("Shutting down worker %u", worker_id);
161       return NULL;
162     }
163   }
164 }
165
166 #ifdef HAVE_FUTEX_H
167 static void futex_wait(int *uaddr, int val)
168 {
169   XBT_VERB("Waiting on futex %p", uaddr);
170   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
171 }
172
173 static void futex_wake(int *uaddr, int val)
174 {
175   XBT_VERB("Waking futex %p", uaddr);
176   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
177 }
178
179 static void xbt_event_init(xbt_event_t event)
180 {
181   int myflag = event->done;
182   if (event->thread_counter < event->threads_to_wait) {
183     futex_wait(&event->done, myflag);
184   }
185 }
186
187 static void xbt_event_signal(xbt_event_t event)
188 {
189   int myflag = event->done;
190   event->thread_counter = 0;
191   event->work++;
192   futex_wake(&event->work, event->threads_to_wait);
193   futex_wait(&event->done, myflag);
194 }
195
196 static void xbt_event_wait(xbt_event_t event)
197 {
198   int myflag;
199   unsigned int mycount;
200
201   myflag = event->work;
202   mycount = __sync_add_and_fetch(&event->thread_counter, 1);
203   if (mycount == event->threads_to_wait) {
204     event->done++;
205     futex_wake(&event->done, 1);
206   }
207
208   futex_wait(&event->work, myflag);
209 }
210
211 static void xbt_event_end(xbt_event_t event)
212 {
213   unsigned int mycount;
214
215   mycount = __sync_add_and_fetch(&event->thread_counter, 1);
216   if (mycount == event->threads_to_wait) {
217     event->done++;
218     futex_wake(&event->done, 1);
219   }
220 }
221 #endif
222
223 #ifdef SIMGRID_TEST
224 #include "xbt.h"
225 #include "xbt/ex.h"
226
227 XBT_TEST_SUITE("parmap", "Parallel Map");
228 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(xbt_parmap_unit);
229
230 xbt_parmap_t parmap;
231
232 void fun(void *arg);
233
234 void fun(void *arg)
235 {
236   //XBT_INFO("I'm job %lu", (unsigned long)arg);
237 }
238
239 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
240 {
241   xbt_test_add("Create the parmap");
242
243   unsigned long i, j;
244   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
245
246   /* Create the parallel map */
247   parmap = xbt_parmap_new(10);
248
249   for(j = 0; j < 100; j++) {
250     xbt_dynar_push_as(data, void *, (void *)j);
251   }
252
253   for (i = 0; i < 5; i++) {
254     xbt_parmap_apply(parmap, fun, data);
255   }
256
257   /* Destroy the parmap */
258   xbt_parmap_destroy(parmap);
259   xbt_dynar_free(&data);
260 }
261
262 #endif /* SIMGRID_TEST */