Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Direct context switching: clean the semantics of parmap
[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 #ifndef _XBT_WIN32
9 #include <sys/syscall.h>
10 #endif
11
12 #ifdef HAVE_FUTEX_H
13         #include <linux/futex.h>
14 #else
15         #include "xbt/xbt_os_thread.h"
16 #endif
17 #include <errno.h>
18 #include "parmap_private.h"
19
20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
21 XBT_LOG_NEW_SUBCATEGORY(xbt_parmap_unit, xbt_parmap, "parmap unit testing");
22
23 static void *_xbt_parmap_worker_main(void *parmap);
24 #ifdef HAVE_FUTEX_H
25         static void futex_wait(int *uaddr, int val);
26         static void futex_wake(int *uaddr, int val);
27 #endif
28 xbt_parmap_t xbt_parmap_new(unsigned int num_workers)
29 {
30   unsigned int i;
31   xbt_os_thread_t worker = NULL;
32
33   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
34
35   /* Initialize the thread pool data structure */
36   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
37 #ifdef HAVE_FUTEX_H
38   parmap->sync_event = xbt_new0(s_xbt_event_t, 1);
39 #endif
40   parmap->num_workers = num_workers;
41   parmap->status = PARMAP_WORK;
42 #ifdef HAVE_FUTEX_H
43   parmap->sync_event->threads_to_wait = num_workers;
44 #endif
45   /* Create the pool of worker threads */
46   for(i=0; i < num_workers; i++){
47     worker = xbt_os_thread_create(NULL, _xbt_parmap_worker_main, parmap, NULL);
48     xbt_os_thread_detach(worker);
49   }
50 #ifdef HAVE_FUTEX_H
51   xbt_event_init(parmap->sync_event);
52 #endif
53   return parmap;
54 }
55
56 void xbt_parmap_destroy(xbt_parmap_t parmap)
57
58   parmap->status = PARMAP_DESTROY;
59 #ifdef HAVE_FUTEX_H
60   xbt_event_signal(parmap->sync_event);
61   xbt_free(parmap->sync_event);
62 #endif
63   xbt_free(parmap);
64 }
65
66  void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
67 {
68   /* Assign resources to worker threads*/
69   parmap->fun = fun;
70   parmap->data = data;
71   parmap->index = 0;
72 #ifdef HAVE_FUTEX_H
73   xbt_event_signal(parmap->sync_event);
74 #endif
75   XBT_DEBUG("Job done");
76 }
77
78 void* xbt_parmap_next(xbt_parmap_t parmap) {
79
80   unsigned int index = __sync_fetch_and_add(&parmap->index, 1);
81   if (index < xbt_dynar_length(parmap->data)) {
82     return xbt_dynar_get_as(parmap->data, index, void*);
83   }
84   return NULL;
85 }
86
87 unsigned long xbt_parmap_get_worker_id(xbt_parmap_t parmap) {
88   return (unsigned long) xbt_os_thread_get_extra_data();
89 }
90
91 static void *_xbt_parmap_worker_main(void *arg)
92 {
93   unsigned int worker_id;
94   xbt_parmap_t parmap = (xbt_parmap_t) arg;
95
96   /* Fetch a worker id */
97   worker_id = __sync_fetch_and_add(&parmap->workers_max_id, 1);
98   xbt_os_thread_set_extra_data((void*) (unsigned long) worker_id);
99
100   XBT_DEBUG("New worker thread created (%u)", worker_id);
101   
102   /* Worker's main loop */
103   while (1) {
104 #ifdef HAVE_FUTEX_H
105     xbt_event_wait(parmap->sync_event);
106 #endif
107     if (parmap->status == PARMAP_WORK) {
108
109       XBT_DEBUG("Worker %u got a job", worker_id);
110
111       void* work = xbt_parmap_next(parmap);
112       if (work != NULL) {
113         parmap->fun(work);
114       }
115
116       XBT_DEBUG("Worker %u has finished", worker_id);
117
118     /* We are destroying the parmap */
119     } else {
120 #ifdef HAVE_FUTEX_H
121       xbt_event_end(parmap->sync_event);
122 #endif
123       XBT_DEBUG("Shutting down worker %u", worker_id);
124       return NULL;
125     }
126   }
127 }
128
129 #ifdef HAVE_FUTEX_H
130 static void futex_wait(int *uaddr, int val)
131 {
132   XBT_VERB("Waiting on futex %p", uaddr);
133   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
134 }
135
136 static void futex_wake(int *uaddr, int val)
137 {
138   XBT_VERB("Waking futex %p", uaddr);
139   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
140 }
141
142 void xbt_event_init(xbt_event_t event)
143 {
144   int myflag = event->done;
145   if(event->thread_counter < event->threads_to_wait)
146     futex_wait(&event->done, myflag);
147 }
148
149 void xbt_event_signal(xbt_event_t event)
150 {
151   int myflag = event->done;
152   event->thread_counter = 0;
153   event->work++;
154   futex_wake(&event->work, event->threads_to_wait);
155   futex_wait(&event->done, myflag);
156 }
157
158 void xbt_event_wait(xbt_event_t event)
159 {
160   int myflag;
161   unsigned int mycount;
162
163   myflag = event->work;
164   mycount = __sync_add_and_fetch(&event->thread_counter, 1);
165   if(mycount == event->threads_to_wait){
166     event->done++;
167     futex_wake(&event->done, 1);
168   }
169
170   futex_wait(&event->work, myflag);
171 }
172
173 void xbt_event_end(xbt_event_t event)
174 {
175   unsigned int mycount;
176
177   mycount = __sync_add_and_fetch(&event->thread_counter, 1);
178   if(mycount == event->threads_to_wait){
179     event->done++;
180     futex_wake(&event->done, 1);
181   }
182 }
183 #endif
184
185 #ifdef SIMGRID_TEST
186 #include "xbt.h"
187 #include "xbt/ex.h"
188
189 XBT_TEST_SUITE("parmap", "Parallel Map");
190 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(xbt_parmap_unit);
191
192
193
194 xbt_parmap_t parmap;
195
196 void fun(void *arg);
197
198 void fun(void *arg)
199 {
200   //XBT_INFO("I'm job %lu", (unsigned long)arg);
201 }
202
203 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
204 {
205   xbt_test_add("Create the parmap");
206
207   unsigned long i,j;
208   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
209
210   /* Create the parallel map */
211   parmap = xbt_parmap_new(10);
212
213   for(j=0; j < 100; j++){
214     xbt_dynar_push_as(data, void *, (void *)j);
215   }
216
217   for(i=0; i < 5; i++)
218     xbt_parmap_apply(parmap, fun, data);
219
220   /* Destroy the parmap */
221   xbt_parmap_destroy(parmap);
222   xbt_dynar_free(&data);
223 }
224
225 #endif /* SIMGRID_TEST */