src/xbt/xbt_os_thread.c

   1 /* xbt_os_thread -- portability layer over the pthread API                  */
   2 /* Used in RL to get win/lin portability, and in SG when CONTEXT_THREAD     */
   3 /* in SG, when using CONTEXT_UCONTEXT, xbt_os_thread_stub is used instead   */
   4
   5 /* Copyright (c) 2007, 2008, 2009, 2010. The SimGrid Team.
   6  * All rights reserved.                                                     */
   7
   8 /* This program is free software; you can redistribute it and/or modify it
   9  * under the terms of the license (GNU LGPL) which comes with this package. */
  10
  11 #include "gras_config.h"
  12 #include "xbt/sysdep.h"
  13 #include "xbt/ex.h"
  14 #include "xbt/ex_interface.h"   /* We play crude games with exceptions */
  15 #include "portable.h"
  16 #include "xbt/xbt_os_time.h"    /* Portable time facilities */
  17 #include "xbt/xbt_os_thread.h"  /* This module */
  18 #include "xbt_modinter.h"       /* Initialization/finalization of this module */
  19
  20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_sync_os, xbt,
  21                                 "Synchronization mechanism (OS-level)");
  22
  23 /* ********************************* PTHREAD IMPLEMENTATION ************************************ */
  24 #ifdef HAVE_PTHREAD_H
  25
  26 #include <pthread.h>
  27 #include <semaphore.h>
  28
  29 #ifdef HAVE_MUTEX_TIMEDLOCK
  30 /* redefine the function header since we fail to get this from system headers on amd (at least) */
  31 int pthread_mutex_timedlock(pthread_mutex_t * mutex,
  32                             const struct timespec *abs_timeout);
  33 #endif
  34
  35
  36 /* use named sempahore when sem_init() does not work */
  37 #ifndef HAVE_SEM_INIT
  38 static int next_sem_ID = 0;
  39 static xbt_os_mutex_t next_sem_ID_lock;
  40 #endif
  41
  42 typedef struct xbt_os_thread_ {
  43   pthread_t t;
  44   int detached;
  45   char *name;
  46   void *param;
  47   pvoid_f_pvoid_t start_routine;
  48   xbt_running_ctx_t *running_ctx;
  49   void *extra_data;
  50 } s_xbt_os_thread_t;
  51 static xbt_os_thread_t main_thread = NULL;
  52
  53 /* thread-specific data containing the xbt_os_thread_t structure */
  54 static pthread_key_t xbt_self_thread_key;
  55 static int thread_mod_inited = 0;
  56
  57 /* frees the xbt_os_thread_t corresponding to the current thread */
  58 static void xbt_os_thread_free_thread_data(xbt_os_thread_t thread)
  59 {
  60   if (thread == main_thread)    /* just killed main thread */
  61     main_thread = NULL;
  62
  63   free(thread->running_ctx);
  64   free(thread->name);
  65   free(thread);
  66 }
  67
  68 /* callback: context fetching */
  69 static xbt_running_ctx_t *_os_thread_get_running_ctx(void)
  70 {
  71   return xbt_os_thread_self()->running_ctx;
  72 }
  73
  74 /* callback: termination */
  75 static void _os_thread_ex_terminate(xbt_ex_t * e)
  76 {
  77   xbt_ex_display(e);
  78
  79   abort();
  80   /* FIXME: there should be a configuration variable to choose to kill everyone or only this one */
  81 }
  82
  83 void xbt_os_thread_mod_preinit(void)
  84 {
  85   int errcode;
  86
  87   if (thread_mod_inited)
  88     return;
  89
  90   if ((errcode = pthread_key_create(&xbt_self_thread_key, NULL)))
  91     THROWF(system_error, errcode,
  92            "pthread_key_create failed for xbt_self_thread_key");
  93
  94   main_thread = xbt_new(s_xbt_os_thread_t, 1);
  95   main_thread->name = (char *) "main";
  96   main_thread->start_routine = NULL;
  97   main_thread->param = NULL;
  98   main_thread->running_ctx = xbt_new(xbt_running_ctx_t, 1);
  99   XBT_RUNNING_CTX_INITIALIZE(main_thread->running_ctx);
 100
 101   if ((errcode = pthread_setspecific(xbt_self_thread_key, main_thread)))
 102     THROWF(system_error, errcode,
 103            "pthread_setspecific failed for xbt_self_thread_key");
 104
 105
 106   __xbt_running_ctx_fetch = _os_thread_get_running_ctx;
 107   __xbt_ex_terminate = _os_thread_ex_terminate;
 108
 109   thread_mod_inited = 1;
 110
 111 #ifndef HAVE_SEM_INIT
 112   next_sem_ID_lock = xbt_os_mutex_init();
 113 #endif
 114
 115 }
 116
 117 void xbt_os_thread_mod_postexit(void)
 118 {
 119   /* FIXME: don't try to free our key on shutdown.
 120      Valgrind detects no leak if we don't, and whine if we try to */
 121   //   int errcode;
 122
 123   //   if ((errcode=pthread_key_delete(xbt_self_thread_key)))
 124   //     THROWF(system_error,errcode,"pthread_key_delete failed for xbt_self_thread_key");
 125   free(main_thread->running_ctx);
 126   free(main_thread);
 127   main_thread = NULL;
 128   thread_mod_inited = 0;
 129 #ifndef HAVE_SEM_INIT
 130   xbt_os_mutex_destroy(next_sem_ID_lock);
 131 #endif
 132
 133   /* Restore the default exception setup */
 134   __xbt_running_ctx_fetch = &__xbt_ex_ctx_default;
 135   __xbt_ex_terminate = &__xbt_ex_terminate_default;
 136 }
 137
 138 int xbt_os_thread_atfork(void (*prepare)(void),
 139                          void (*parent)(void), void (*child)(void))
 140 {
 141   return pthread_atfork(prepare, parent, child);
 142 }
 143
 144 static void *wrapper_start_routine(void *s)
 145 {
 146   xbt_os_thread_t t = s;
 147   int errcode;
 148
 149   if ((errcode = pthread_setspecific(xbt_self_thread_key, t)))
 150     THROWF(system_error, errcode,
 151            "pthread_setspecific failed for xbt_self_thread_key");
 152
 153   void *res = (*(t->start_routine)) (t->param);
 154   if (t->detached)
 155     xbt_os_thread_free_thread_data(t);
 156   return res;
 157 }
 158
 159 xbt_os_thread_t xbt_os_thread_create(const char *name,
 160                                      pvoid_f_pvoid_t start_routine,
 161                                      void *param,
 162                                      void *extra_data)
 163 {
 164   int errcode;
 165
 166   xbt_os_thread_t res_thread = xbt_new(s_xbt_os_thread_t, 1);
 167   res_thread->detached = 0;
 168   res_thread->name = xbt_strdup(name);
 169   res_thread->start_routine = start_routine;
 170   res_thread->param = param;
 171   res_thread->running_ctx = xbt_new(xbt_running_ctx_t, 1);
 172   XBT_RUNNING_CTX_INITIALIZE(res_thread->running_ctx);
 173   res_thread->extra_data = extra_data;
 174
 175   if ((errcode = pthread_create(&(res_thread->t), NULL,
 176                                 wrapper_start_routine, res_thread)))
 177     THROWF(system_error, errcode,
 178            "pthread_create failed: %s", strerror(errcode));
 179
 180   return res_thread;
 181 }
 182
 183 const char *xbt_os_thread_name(xbt_os_thread_t t)
 184 {
 185   return t->name;
 186 }
 187
 188 const char *xbt_os_thread_self_name(void)
 189 {
 190   xbt_os_thread_t me = xbt_os_thread_self();
 191   return me ? me->name : "main";
 192 }
 193
 194 void xbt_os_thread_join(xbt_os_thread_t thread, void **thread_return)
 195 {
 196
 197   int errcode;
 198
 199   if ((errcode = pthread_join(thread->t, thread_return)))
 200     THROWF(system_error, errcode, "pthread_join failed: %s",
 201            strerror(errcode));
 202   xbt_os_thread_free_thread_data(thread);
 203 }
 204
 205 void xbt_os_thread_exit(int *retval)
 206 {
 207   pthread_exit(retval);
 208 }
 209
 210 xbt_os_thread_t xbt_os_thread_self(void)
 211 {
 212   xbt_os_thread_t res;
 213
 214   if (!thread_mod_inited)
 215     return NULL;
 216
 217   res = pthread_getspecific(xbt_self_thread_key);
 218
 219   return res;
 220 }
 221
 222 void xbt_os_thread_detach(xbt_os_thread_t thread)
 223 {
 224   thread->detached = 1;
 225   pthread_detach(thread->t);
 226 }
 227
 228 #include <sched.h>
 229 void xbt_os_thread_yield(void)
 230 {
 231   sched_yield();
 232 }
 233
 234 void xbt_os_thread_cancel(xbt_os_thread_t t)
 235 {
 236   pthread_cancel(t->t);
 237 }
 238
 239 /****** mutex related functions ******/
 240 typedef struct xbt_os_mutex_ {
 241   /* KEEP IT IN SYNC WITH xbt_thread.c */
 242   pthread_mutex_t m;
 243 } s_xbt_os_mutex_t;
 244
 245 typedef struct xbt_os_rmutex_ {
 246   xbt_os_mutex_t mutex;
 247   xbt_os_thread_t owner;
 248   int count;
 249 } s_xbt_os_rmutex_t;
 250
 251 #include <time.h>
 252 #include <math.h>
 253
 254 xbt_os_mutex_t xbt_os_mutex_init(void)
 255 {
 256   xbt_os_mutex_t res = xbt_new(s_xbt_os_mutex_t, 1);
 257   int errcode;
 258
 259   if ((errcode = pthread_mutex_init(&(res->m), NULL)))
 260     THROWF(system_error, errcode, "pthread_mutex_init() failed: %s",
 261            strerror(errcode));
 262
 263   return res;
 264 }
 265
 266 void xbt_os_mutex_acquire(xbt_os_mutex_t mutex)
 267 {
 268   int errcode;
 269
 270   if ((errcode = pthread_mutex_lock(&(mutex->m))))
 271     THROWF(system_error, errcode, "pthread_mutex_lock(%p) failed: %s",
 272            mutex, strerror(errcode));
 273 }
 274
 275
 276 void xbt_os_mutex_timedacquire(xbt_os_mutex_t mutex, double delay)
 277 {
 278   int errcode;
 279
 280   if (delay < 0) {
 281     xbt_os_mutex_acquire(mutex);
 282
 283   } else if (delay == 0) {
 284     errcode = pthread_mutex_trylock(&(mutex->m));
 285
 286     switch (errcode) {
 287     case 0:
 288       return;
 289     case ETIMEDOUT:
 290       THROWF(timeout_error, 0, "mutex %p not ready", mutex);
 291     default:
 292       THROWF(system_error, errcode,
 293              "xbt_mutex_timedacquire(%p) failed: %s", mutex,
 294              strerror(errcode));
 295     }
 296
 297
 298   } else {
 299
 300 #ifdef HAVE_MUTEX_TIMEDLOCK
 301     struct timespec ts_end;
 302     double end = delay + xbt_os_time();
 303
 304     ts_end.tv_sec = (time_t) floor(end);
 305     ts_end.tv_nsec = (long) ((end - ts_end.tv_sec) * 1000000000);
 306     XBT_DEBUG("pthread_mutex_timedlock(%p,%p)", &(mutex->m), &ts_end);
 307
 308     errcode = pthread_mutex_timedlock(&(mutex->m), &ts_end);
 309
 310 #else                           /* Well, let's reimplement it since those lazy libc dudes didn't */
 311     double start = xbt_os_time();
 312     do {
 313       errcode = pthread_mutex_trylock(&(mutex->m));
 314       if (errcode == EBUSY)
 315         xbt_os_thread_yield();
 316     } while (errcode == EBUSY && xbt_os_time() - start < delay);
 317
 318     if (errcode == EBUSY)
 319       errcode = ETIMEDOUT;
 320
 321 #endif                          /* HAVE_MUTEX_TIMEDLOCK */
 322
 323     switch (errcode) {
 324     case 0:
 325       return;
 326
 327     case ETIMEDOUT:
 328       THROWF(timeout_error, delay,
 329              "mutex %p wasn't signaled before timeout (%f)", mutex, delay);
 330
 331     default:
 332       THROWF(system_error, errcode,
 333              "pthread_mutex_timedlock(%p,%f) failed: %s", mutex, delay,
 334              strerror(errcode));
 335     }
 336   }
 337 }
 338
 339 void xbt_os_mutex_release(xbt_os_mutex_t mutex)
 340 {
 341   int errcode;
 342
 343   if ((errcode = pthread_mutex_unlock(&(mutex->m))))
 344     THROWF(system_error, errcode, "pthread_mutex_unlock(%p) failed: %s",
 345            mutex, strerror(errcode));
 346 }
 347
 348 void xbt_os_mutex_destroy(xbt_os_mutex_t mutex)
 349 {
 350   int errcode;
 351
 352   if (!mutex)
 353     return;
 354
 355   if ((errcode = pthread_mutex_destroy(&(mutex->m))))
 356     THROWF(system_error, errcode, "pthread_mutex_destroy(%p) failed: %s",
 357            mutex, strerror(errcode));
 358   free(mutex);
 359 }
 360
 361 xbt_os_rmutex_t xbt_os_rmutex_init(void)
 362 {
 363   xbt_os_rmutex_t rmutex = xbt_new0(struct xbt_os_rmutex_, 0);
 364   rmutex->mutex = xbt_os_mutex_init();
 365   rmutex->owner = NULL;
 366   rmutex->count = 0;
 367   return rmutex;
 368 }
 369
 370 void xbt_os_rmutex_acquire(xbt_os_rmutex_t rmutex)
 371 {
 372   xbt_os_thread_t self = xbt_os_thread_self();
 373   xbt_assert(self != NULL, "Cannot get my own thread object (is the thread module initialized?)");
 374
 375   if (self != rmutex->owner) {
 376     xbt_os_mutex_acquire(rmutex->mutex);
 377     rmutex->owner = self;
 378     rmutex->count = 1;
 379   } else {
 380     rmutex->count++;
 381  }
 382 }
 383
 384 void xbt_os_rmutex_release(xbt_os_rmutex_t rmutex)
 385 {
 386   xbt_assert(rmutex->owner == xbt_os_thread_self());
 387
 388   if (--rmutex->count == 0) {
 389     rmutex->owner = NULL;
 390     xbt_os_mutex_release(rmutex->mutex);
 391   }
 392 }
 393
 394 void xbt_os_rmutex_destroy(xbt_os_rmutex_t rmutex)
 395 {
 396   xbt_os_mutex_destroy(rmutex->mutex);
 397   xbt_free(rmutex);
 398 }
 399
 400 /***** condition related functions *****/
 401 typedef struct xbt_os_cond_ {
 402   /* KEEP IT IN SYNC WITH xbt_thread.c */
 403   pthread_cond_t c;
 404 } s_xbt_os_cond_t;
 405
 406 xbt_os_cond_t xbt_os_cond_init(void)
 407 {
 408   xbt_os_cond_t res = xbt_new(s_xbt_os_cond_t, 1);
 409   int errcode;
 410   if ((errcode = pthread_cond_init(&(res->c), NULL)))
 411     THROWF(system_error, errcode, "pthread_cond_init() failed: %s",
 412            strerror(errcode));
 413
 414   return res;
 415 }
 416
 417 void xbt_os_cond_wait(xbt_os_cond_t cond, xbt_os_mutex_t mutex)
 418 {
 419   int errcode;
 420   if ((errcode = pthread_cond_wait(&(cond->c), &(mutex->m))))
 421     THROWF(system_error, errcode, "pthread_cond_wait(%p,%p) failed: %s",
 422            cond, mutex, strerror(errcode));
 423 }
 424
 425
 426 void xbt_os_cond_timedwait(xbt_os_cond_t cond, xbt_os_mutex_t mutex,
 427                            double delay)
 428 {
 429   int errcode;
 430   struct timespec ts_end;
 431   double end = delay + xbt_os_time();
 432
 433   if (delay < 0) {
 434     xbt_os_cond_wait(cond, mutex);
 435   } else {
 436     ts_end.tv_sec = (time_t) floor(end);
 437     ts_end.tv_nsec = (long) ((end - ts_end.tv_sec) * 1000000000);
 438     XBT_DEBUG("pthread_cond_timedwait(%p,%p,%p)", &(cond->c), &(mutex->m),
 439            &ts_end);
 440     switch ((errcode =
 441              pthread_cond_timedwait(&(cond->c), &(mutex->m), &ts_end))) {
 442     case 0:
 443       return;
 444     case ETIMEDOUT:
 445       THROWF(timeout_error, errcode,
 446              "condition %p (mutex %p) wasn't signaled before timeout (%f)",
 447              cond, mutex, delay);
 448     default:
 449       THROWF(system_error, errcode,
 450              "pthread_cond_timedwait(%p,%p,%f) failed: %s", cond, mutex,
 451              delay, strerror(errcode));
 452     }
 453   }
 454 }
 455
 456 void xbt_os_cond_signal(xbt_os_cond_t cond)
 457 {
 458   int errcode;
 459   if ((errcode = pthread_cond_signal(&(cond->c))))
 460     THROWF(system_error, errcode, "pthread_cond_signal(%p) failed: %s",
 461            cond, strerror(errcode));
 462 }
 463
 464 void xbt_os_cond_broadcast(xbt_os_cond_t cond)
 465 {
 466   int errcode;
 467   if ((errcode = pthread_cond_broadcast(&(cond->c))))
 468     THROWF(system_error, errcode, "pthread_cond_broadcast(%p) failed: %s",
 469            cond, strerror(errcode));
 470 }
 471
 472 void xbt_os_cond_destroy(xbt_os_cond_t cond)
 473 {
 474   int errcode;
 475
 476   if (!cond)
 477     return;
 478
 479   if ((errcode = pthread_cond_destroy(&(cond->c))))
 480     THROWF(system_error, errcode, "pthread_cond_destroy(%p) failed: %s",
 481            cond, strerror(errcode));
 482   free(cond);
 483 }
 484
 485 void *xbt_os_thread_getparam(void)
 486 {
 487   xbt_os_thread_t t = xbt_os_thread_self();
 488   return t ? t->param : NULL;
 489 }
 490
 491 typedef struct xbt_os_sem_ {
 492 #ifndef HAVE_SEM_INIT
 493   char *name;
 494 #endif
 495   sem_t s;
 496   sem_t *ps;
 497 } s_xbt_os_sem_t;
 498
 499 #ifndef SEM_FAILED
 500 #define SEM_FAILED (-1)
 501 #endif
 502
 503 xbt_os_sem_t xbt_os_sem_init(unsigned int value)
 504 {
 505   xbt_os_sem_t res = xbt_new(s_xbt_os_sem_t, 1);
 506
 507   /* On some systems (MAC OS X), only the stub of sem_init is to be found.
 508    * Any attempt to use it leads to ENOSYS (function not implemented).
 509    * If such a prehistoric system is detected, do the job with sem_open instead
 510    */
 511 #ifdef HAVE_SEM_INIT
 512   if (sem_init(&(res->s), 0, value) != 0)
 513     THROWF(system_error, errno, "sem_init() failed: %s", strerror(errno));
 514   res->ps = &(res->s);
 515
 516 #else                           /* damn, no sem_init(). Reimplement it */
 517
 518   xbt_os_mutex_acquire(next_sem_ID_lock);
 519   res->name = bprintf("/%d", ++next_sem_ID);
 520   xbt_os_mutex_release(next_sem_ID_lock);
 521
 522   res->ps = sem_open(res->name, O_CREAT, 0644, value);
 523   if ((res->ps == (sem_t *) SEM_FAILED) && (errno == ENAMETOOLONG)) {
 524     /* Old darwins only allow 13 chars. Did you create *that* amount of semaphores? */
 525     res->name[13] = '\0';
 526     res->ps = sem_open(res->name, O_CREAT, 0644, 1);
 527   }
 528   if ((res->ps == (sem_t *) SEM_FAILED))
 529     THROWF(system_error, errno, "sem_open() failed: %s", strerror(errno));
 530
 531   /* Remove the name from the semaphore namespace: we never join on it */
 532   if (sem_unlink(res->name) < 0)
 533     THROWF(system_error, errno, "sem_unlink() failed: %s",
 534            strerror(errno));
 535
 536 #endif
 537
 538   return res;
 539 }
 540
 541 void xbt_os_sem_acquire(xbt_os_sem_t sem)
 542 {
 543   if (!sem)
 544     THROWF(arg_error, EINVAL, "Cannot acquire of the NULL semaphore");
 545   if (sem_wait(sem->ps) < 0)
 546     THROWF(system_error, errno, "sem_wait() failed: %s", strerror(errno));
 547 }
 548
 549 void xbt_os_sem_timedacquire(xbt_os_sem_t sem, double delay)
 550 {
 551   int errcode;
 552
 553   if (!sem)
 554     THROWF(arg_error, EINVAL, "Cannot acquire of the NULL semaphore");
 555
 556   if (delay < 0) {
 557     xbt_os_sem_acquire(sem);
 558   } else if (delay == 0) {
 559     errcode = sem_trywait(sem->ps);
 560
 561     switch (errcode) {
 562     case 0:
 563       return;
 564     case ETIMEDOUT:
 565       THROWF(timeout_error, 0, "semaphore %p not ready", sem);
 566     default:
 567       THROWF(system_error, errcode,
 568              "xbt_os_sem_timedacquire(%p) failed: %s", sem,
 569              strerror(errcode));
 570     }
 571
 572   } else {
 573 #ifdef HAVE_SEM_WAIT
 574     struct timespec ts_end;
 575     double end = delay + xbt_os_time();
 576
 577     ts_end.tv_sec = (time_t) floor(end);
 578     ts_end.tv_nsec = (long) ((end - ts_end.tv_sec) * 1000000000);
 579     XBT_DEBUG("sem_timedwait(%p,%p)", sem->ps, &ts_end);
 580     errcode = sem_timedwait(sem->s, &ts_end);
 581
 582 #else                           /* Okay, reimplement this function then */
 583     double start = xbt_os_time();
 584     do {
 585       errcode = sem_trywait(sem->ps);
 586       if (errcode == EBUSY)
 587         xbt_os_thread_yield();
 588     } while (errcode == EBUSY && xbt_os_time() - start < delay);
 589
 590     if (errcode == EBUSY)
 591       errcode = ETIMEDOUT;
 592 #endif
 593
 594     switch (errcode) {
 595     case 0:
 596       return;
 597
 598     case ETIMEDOUT:
 599       THROWF(timeout_error, delay,
 600              "semaphore %p wasn't signaled before timeout (%f)", sem,
 601              delay);
 602
 603     default:
 604       THROWF(system_error, errcode, "sem_timedwait(%p,%f) failed: %s", sem,
 605              delay, strerror(errcode));
 606     }
 607   }
 608 }
 609
 610 void xbt_os_sem_release(xbt_os_sem_t sem)
 611 {
 612   if (!sem)
 613     THROWF(arg_error, EINVAL, "Cannot release of the NULL semaphore");
 614
 615   if (sem_post(sem->ps) < 0)
 616     THROWF(system_error, errno, "sem_post() failed: %s", strerror(errno));
 617 }
 618
 619 void xbt_os_sem_destroy(xbt_os_sem_t sem)
 620 {
 621   if (!sem)
 622     THROWF(arg_error, EINVAL, "Cannot destroy the NULL sempahore");
 623
 624 #ifdef HAVE_SEM_INIT
 625   if (sem_destroy(sem->ps) < 0)
 626     THROWF(system_error, errno, "sem_destroy() failed: %s",
 627            strerror(errno));
 628 #else
 629   if (sem_close(sem->ps) < 0)
 630     THROWF(system_error, errno, "sem_close() failed: %s", strerror(errno));
 631   xbt_free(sem->name);
 632
 633 #endif
 634   xbt_free(sem);
 635 }
 636
 637 void xbt_os_sem_get_value(xbt_os_sem_t sem, int *svalue)
 638 {
 639   if (!sem)
 640     THROWF(arg_error, EINVAL,
 641            "Cannot get the value of the NULL semaphore");
 642
 643   if (sem_getvalue(&(sem->s), svalue) < 0)
 644     THROWF(system_error, errno, "sem_getvalue() failed: %s",
 645            strerror(errno));
 646 }
 647
 648 /* ********************************* WINDOWS IMPLEMENTATION ************************************ */
 649
 650 #elif defined(_XBT_WIN32)
 651
 652 #include <math.h>
 653
 654 typedef struct xbt_os_thread_ {
 655   char *name;
 656   HANDLE handle;                /* the win thread handle        */
 657   unsigned long id;             /* the win thread id            */
 658   pvoid_f_pvoid_t start_routine;
 659   void *param;
 660   void *extra_data;
 661 } s_xbt_os_thread_t;
 662
 663 /* so we can specify the size of the stack of the threads */
 664 #ifndef STACK_SIZE_PARAM_IS_A_RESERVATION
 665 #define STACK_SIZE_PARAM_IS_A_RESERVATION 0x00010000
 666 #endif
 667
 668 /* the default size of the stack of the threads (in bytes)*/
 669 #define XBT_DEFAULT_THREAD_STACK_SIZE   4096
 670
 671 /* key to the TLS containing the xbt_os_thread_t structure */
 672 static unsigned long xbt_self_thread_key;
 673
 674 void xbt_os_thread_mod_preinit(void)
 675 {
 676   xbt_self_thread_key = TlsAlloc();
 677 }
 678
 679 void xbt_os_thread_mod_postexit(void)
 680 {
 681
 682   if (!TlsFree(xbt_self_thread_key))
 683     THROWF(system_error, (int) GetLastError(),
 684            "TlsFree() failed to cleanup the thread submodule");
 685 }
 686
 687 int xbt_os_thread_atfork(void (*prepare)(void),
 688                          void (*parent)(void), void (*child)(void))
 689 {
 690   return 0;
 691 }
 692
 693 static DWORD WINAPI wrapper_start_routine(void *s)
 694 {
 695   xbt_os_thread_t t = (xbt_os_thread_t) s;
 696   DWORD *rv;
 697
 698   if (!TlsSetValue(xbt_self_thread_key, t))
 699     THROWF(system_error, (int) GetLastError(),
 700            "TlsSetValue of data describing the created thread failed");
 701
 702   rv = (DWORD *) ((t->start_routine) (t->param));
 703
 704   return rv ? *rv : 0;
 705
 706 }
 707
 708
 709 xbt_os_thread_t xbt_os_thread_create(const char *name,
 710                                      pvoid_f_pvoid_t start_routine,
 711                                      void *param,
 712                                      void *extra_data)
 713 {
 714
 715   xbt_os_thread_t t = xbt_new(s_xbt_os_thread_t, 1);
 716
 717   t->name = xbt_strdup(name);
 718   t->start_routine = start_routine;
 719   t->param = param;
 720   t->extra_data = extra_data;
 721   t->handle = CreateThread(NULL, XBT_DEFAULT_THREAD_STACK_SIZE,
 722                            (LPTHREAD_START_ROUTINE) wrapper_start_routine,
 723                            t, STACK_SIZE_PARAM_IS_A_RESERVATION, &(t->id));
 724
 725   if (!t->handle) {
 726     xbt_free(t);
 727     THROWF(system_error, (int) GetLastError(), "CreateThread failed");
 728   }
 729
 730   return t;
 731 }
 732
 733 const char *xbt_os_thread_name(xbt_os_thread_t t)
 734 {
 735   return t->name;
 736 }
 737
 738 const char *xbt_os_thread_self_name(void)
 739 {
 740   xbt_os_thread_t t = xbt_os_thread_self();
 741   return t ? t->name : "main";
 742 }
 743
 744 void xbt_os_thread_join(xbt_os_thread_t thread, void **thread_return)
 745 {
 746
 747   if (WAIT_OBJECT_0 != WaitForSingleObject(thread->handle, INFINITE))
 748     THROWF(system_error, (int) GetLastError(),
 749            "WaitForSingleObject failed");
 750
 751   if (thread_return) {
 752
 753     if (!GetExitCodeThread(thread->handle, (DWORD *) (*thread_return)))
 754       THROWF(system_error, (int) GetLastError(),
 755              "GetExitCodeThread failed");
 756   }
 757
 758   CloseHandle(thread->handle);
 759
 760   if (thread->name)
 761     free(thread->name);
 762
 763   free(thread);
 764 }
 765
 766 void xbt_os_thread_exit(int *retval)
 767 {
 768   if (retval)
 769     ExitThread(*retval);
 770   else
 771     ExitThread(0);
 772 }
 773
 774 void xbt_os_thread_detach(xbt_os_thread_t thread)
 775 {
 776   THROW_UNIMPLEMENTED;
 777 }
 778
 779
 780 xbt_os_thread_t xbt_os_thread_self(void)
 781 {
 782   return TlsGetValue(xbt_self_thread_key);
 783 }
 784
 785 void *xbt_os_thread_getparam(void)
 786 {
 787   xbt_os_thread_t t = xbt_os_thread_self();
 788   return t->param;
 789 }
 790
 791
 792 void xbt_os_thread_yield(void)
 793 {
 794   Sleep(0);
 795 }
 796
 797 void xbt_os_thread_cancel(xbt_os_thread_t t)
 798 {
 799   if (!TerminateThread(t->handle, 0))
 800     THROWF(system_error, (int) GetLastError(), "TerminateThread failed");
 801 }
 802
 803 /****** mutex related functions ******/
 804 typedef struct xbt_os_mutex_ {
 805   /* KEEP IT IN SYNC WITH xbt_thread.c */
 806   CRITICAL_SECTION lock;
 807 } s_xbt_os_mutex_t;
 808
 809 xbt_os_mutex_t xbt_os_mutex_init(void)
 810 {
 811   xbt_os_mutex_t res = xbt_new(s_xbt_os_mutex_t, 1);
 812
 813   /* initialize the critical section object */
 814   InitializeCriticalSection(&(res->lock));
 815
 816   return res;
 817 }
 818
 819 void xbt_os_mutex_acquire(xbt_os_mutex_t mutex)
 820 {
 821   EnterCriticalSection(&mutex->lock);
 822 }
 823
 824 void xbt_os_mutex_timedacquire(xbt_os_mutex_t mutex, double delay)
 825 {
 826   THROW_UNIMPLEMENTED;
 827 }
 828
 829 void xbt_os_mutex_release(xbt_os_mutex_t mutex)
 830 {
 831
 832   LeaveCriticalSection(&mutex->lock);
 833
 834 }
 835
 836 void xbt_os_mutex_destroy(xbt_os_mutex_t mutex)
 837 {
 838
 839   if (!mutex)
 840     return;
 841
 842   DeleteCriticalSection(&mutex->lock);
 843   free(mutex);
 844 }
 845
 846 /***** condition related functions *****/
 847 enum {                          /* KEEP IT IN SYNC WITH xbt_thread.c */
 848   SIGNAL = 0,
 849   BROADCAST = 1,
 850   MAX_EVENTS = 2
 851 };
 852
 853 typedef struct xbt_os_cond_ {
 854   /* KEEP IT IN SYNC WITH xbt_thread.c */
 855   HANDLE events[MAX_EVENTS];
 856
 857   unsigned int waiters_count;   /* the number of waiters                        */
 858   CRITICAL_SECTION waiters_count_lock;  /* protect access to waiters_count  */
 859 } s_xbt_os_cond_t;
 860
 861 xbt_os_cond_t xbt_os_cond_init(void)
 862 {
 863
 864   xbt_os_cond_t res = xbt_new0(s_xbt_os_cond_t, 1);
 865
 866   memset(&res->waiters_count_lock, 0, sizeof(CRITICAL_SECTION));
 867
 868   /* initialize the critical section object */
 869   InitializeCriticalSection(&res->waiters_count_lock);
 870
 871   res->waiters_count = 0;
 872
 873   /* Create an auto-reset event */
 874   res->events[SIGNAL] = CreateEvent(NULL, FALSE, FALSE, NULL);
 875
 876   if (!res->events[SIGNAL]) {
 877     DeleteCriticalSection(&res->waiters_count_lock);
 878     free(res);
 879     THROWF(system_error, 0, "CreateEvent failed for the signals");
 880   }
 881
 882   /* Create a manual-reset event. */
 883   res->events[BROADCAST] = CreateEvent(NULL, TRUE, FALSE, NULL);
 884
 885   if (!res->events[BROADCAST]) {
 886
 887     DeleteCriticalSection(&res->waiters_count_lock);
 888     CloseHandle(res->events[SIGNAL]);
 889     free(res);
 890     THROWF(system_error, 0, "CreateEvent failed for the broadcasts");
 891   }
 892
 893   return res;
 894 }
 895
 896 void xbt_os_cond_wait(xbt_os_cond_t cond, xbt_os_mutex_t mutex)
 897 {
 898
 899   unsigned long wait_result;
 900   int is_last_waiter;
 901
 902   /* lock the threads counter and increment it */
 903   EnterCriticalSection(&cond->waiters_count_lock);
 904   cond->waiters_count++;
 905   LeaveCriticalSection(&cond->waiters_count_lock);
 906
 907   /* unlock the mutex associate with the condition */
 908   LeaveCriticalSection(&mutex->lock);
 909
 910   /* wait for a signal (broadcast or no) */
 911   wait_result = WaitForMultipleObjects(2, cond->events, FALSE, INFINITE);
 912
 913   if (wait_result == WAIT_FAILED)
 914     THROWF(system_error, 0,
 915            "WaitForMultipleObjects failed, so we cannot wait on the condition");
 916
 917   /* we have a signal lock the condition */
 918   EnterCriticalSection(&cond->waiters_count_lock);
 919   cond->waiters_count--;
 920
 921   /* it's the last waiter or it's a broadcast ? */
 922   is_last_waiter = ((wait_result == WAIT_OBJECT_0 + BROADCAST - 1)
 923                     && (cond->waiters_count == 0));
 924
 925   LeaveCriticalSection(&cond->waiters_count_lock);
 926
 927   /* yes it's the last waiter or it's a broadcast
 928    * only reset the manual event (the automatic event is reset in the WaitForMultipleObjects() function
 929    * by the system.
 930    */
 931   if (is_last_waiter)
 932     if (!ResetEvent(cond->events[BROADCAST]))
 933       THROWF(system_error, 0, "ResetEvent failed");
 934
 935   /* relock the mutex associated with the condition in accordance with the posix thread specification */
 936   EnterCriticalSection(&mutex->lock);
 937 }
 938
 939 void xbt_os_cond_timedwait(xbt_os_cond_t cond, xbt_os_mutex_t mutex,
 940                            double delay)
 941 {
 942
 943   unsigned long wait_result = WAIT_TIMEOUT;
 944   int is_last_waiter;
 945   unsigned long end = (unsigned long) (delay * 1000);
 946
 947
 948   if (delay < 0) {
 949     xbt_os_cond_wait(cond, mutex);
 950   } else {
 951     XBT_DEBUG("xbt_cond_timedwait(%p,%p,%lu)", &(cond->events),
 952            &(mutex->lock), end);
 953
 954     /* lock the threads counter and increment it */
 955     EnterCriticalSection(&cond->waiters_count_lock);
 956     cond->waiters_count++;
 957     LeaveCriticalSection(&cond->waiters_count_lock);
 958
 959     /* unlock the mutex associate with the condition */
 960     LeaveCriticalSection(&mutex->lock);
 961     /* wait for a signal (broadcast or no) */
 962
 963     wait_result = WaitForMultipleObjects(2, cond->events, FALSE, end);
 964
 965     switch (wait_result) {
 966     case WAIT_TIMEOUT:
 967       THROWF(timeout_error, GetLastError(),
 968              "condition %p (mutex %p) wasn't signaled before timeout (%f)",
 969              cond, mutex, delay);
 970     case WAIT_FAILED:
 971       THROWF(system_error, GetLastError(),
 972              "WaitForMultipleObjects failed, so we cannot wait on the condition");
 973     }
 974
 975     /* we have a signal lock the condition */
 976     EnterCriticalSection(&cond->waiters_count_lock);
 977     cond->waiters_count--;
 978
 979     /* it's the last waiter or it's a broadcast ? */
 980     is_last_waiter = ((wait_result == WAIT_OBJECT_0 + BROADCAST - 1)
 981                       && (cond->waiters_count == 0));
 982
 983     LeaveCriticalSection(&cond->waiters_count_lock);
 984
 985     /* yes it's the last waiter or it's a broadcast
 986      * only reset the manual event (the automatic event is reset in the WaitForMultipleObjects() function
 987      * by the system.
 988      */
 989     if (is_last_waiter)
 990       if (!ResetEvent(cond->events[BROADCAST]))
 991         THROWF(system_error, 0, "ResetEvent failed");
 992
 993     /* relock the mutex associated with the condition in accordance with the posix thread specification */
 994     EnterCriticalSection(&mutex->lock);
 995   }
 996   /*THROW_UNIMPLEMENTED; */
 997 }
 998
 999 void xbt_os_cond_signal(xbt_os_cond_t cond)
1000 {
1001   int have_waiters;
1002
1003   EnterCriticalSection(&cond->waiters_count_lock);
1004   have_waiters = cond->waiters_count > 0;
1005   LeaveCriticalSection(&cond->waiters_count_lock);
1006
1007   if (have_waiters)
1008     if (!SetEvent(cond->events[SIGNAL]))
1009       THROWF(system_error, 0, "SetEvent failed");
1010
1011   xbt_os_thread_yield();
1012 }
1013
1014 void xbt_os_cond_broadcast(xbt_os_cond_t cond)
1015 {
1016   int have_waiters;
1017
1018   EnterCriticalSection(&cond->waiters_count_lock);
1019   have_waiters = cond->waiters_count > 0;
1020   LeaveCriticalSection(&cond->waiters_count_lock);
1021
1022   if (have_waiters)
1023     SetEvent(cond->events[BROADCAST]);
1024 }
1025
1026 void xbt_os_cond_destroy(xbt_os_cond_t cond)
1027 {
1028   int error = 0;
1029
1030   if (!cond)
1031     return;
1032
1033   if (!CloseHandle(cond->events[SIGNAL]))
1034     error = 1;
1035
1036   if (!CloseHandle(cond->events[BROADCAST]))
1037     error = 1;
1038
1039   DeleteCriticalSection(&cond->waiters_count_lock);
1040
1041   xbt_free(cond);
1042
1043   if (error)
1044     THROWF(system_error, 0, "Error while destroying the condition");
1045 }
1046
1047 typedef struct xbt_os_sem_ {
1048   HANDLE h;
1049   unsigned int value;
1050   CRITICAL_SECTION value_lock;  /* protect access to value of the semaphore  */
1051 } s_xbt_os_sem_t;
1052
1053 #ifndef INT_MAX
1054 # define INT_MAX 32767          /* let's be safe by underestimating this value: this is for 16bits only */
1055 #endif
1056
1057 xbt_os_sem_t xbt_os_sem_init(unsigned int value)
1058 {
1059   xbt_os_sem_t res;
1060
1061   if (value > INT_MAX)
1062     THROWF(arg_error, value,
1063            "Semaphore initial value too big: %ud cannot be stored as a signed int",
1064            value);
1065
1066   res = (xbt_os_sem_t) xbt_new0(s_xbt_os_sem_t, 1);
1067
1068   if (!(res->h = CreateSemaphore(NULL, value, (long) INT_MAX, NULL))) {
1069     THROWF(system_error, GetLastError(), "CreateSemaphore() failed: %s",
1070            strerror(GetLastError()));
1071     return NULL;
1072   }
1073
1074   res->value = value;
1075
1076   InitializeCriticalSection(&(res->value_lock));
1077
1078   return res;
1079 }
1080
1081 void xbt_os_sem_acquire(xbt_os_sem_t sem)
1082 {
1083   if (!sem)
1084     THROWF(arg_error, EINVAL, "Cannot acquire the NULL semaphore");
1085
1086   /* wait failure */
1087   if (WAIT_OBJECT_0 != WaitForSingleObject(sem->h, INFINITE))
1088     THROWF(system_error, GetLastError(),
1089            "WaitForSingleObject() failed: %s", strerror(GetLastError()));
1090   EnterCriticalSection(&(sem->value_lock));
1091   sem->value--;
1092   LeaveCriticalSection(&(sem->value_lock));
1093 }
1094
1095 void xbt_os_sem_timedacquire(xbt_os_sem_t sem, double timeout)
1096 {
1097   long seconds;
1098   long milliseconds;
1099   double end = timeout + xbt_os_time();
1100
1101   if (!sem)
1102     THROWF(arg_error, EINVAL, "Cannot acquire the NULL semaphore");
1103
1104   if (timeout < 0) {
1105     xbt_os_sem_acquire(sem);
1106   } else {                      /* timeout can be zero <-> try acquire ) */
1107
1108
1109     seconds = (long) floor(end);
1110     milliseconds = (long) ((end - seconds) * 1000);
1111     milliseconds += (seconds * 1000);
1112
1113     switch (WaitForSingleObject(sem->h, milliseconds)) {
1114     case WAIT_OBJECT_0:
1115       EnterCriticalSection(&(sem->value_lock));
1116       sem->value--;
1117       LeaveCriticalSection(&(sem->value_lock));
1118       return;
1119
1120     case WAIT_TIMEOUT:
1121       THROWF(timeout_error, GetLastError(),
1122              "semaphore %p wasn't signaled before timeout (%f)", sem,
1123              timeout);
1124       return;
1125
1126     default:
1127       THROWF(system_error, GetLastError(),
1128              "WaitForSingleObject(%p,%f) failed: %s", sem, timeout,
1129              strerror(GetLastError()));
1130     }
1131   }
1132 }
1133
1134 void xbt_os_sem_release(xbt_os_sem_t sem)
1135 {
1136   if (!sem)
1137     THROWF(arg_error, EINVAL, "Cannot release the NULL semaphore");
1138
1139   if (!ReleaseSemaphore(sem->h, 1, NULL))
1140     THROWF(system_error, GetLastError(), "ReleaseSemaphore() failed: %s",
1141            strerror(GetLastError()));
1142   EnterCriticalSection(&(sem->value_lock));
1143   sem->value++;
1144   LeaveCriticalSection(&(sem->value_lock));
1145 }
1146
1147 void xbt_os_sem_destroy(xbt_os_sem_t sem)
1148 {
1149   if (!sem)
1150     THROWF(arg_error, EINVAL, "Cannot destroy the NULL semaphore");
1151
1152   if (!CloseHandle(sem->h))
1153     THROWF(system_error, GetLastError(), "CloseHandle() failed: %s",
1154            strerror(GetLastError()));
1155
1156   DeleteCriticalSection(&(sem->value_lock));
1157
1158   xbt_free(sem);
1159
1160 }
1161
1162 void xbt_os_sem_get_value(xbt_os_sem_t sem, int *svalue)
1163 {
1164   if (!sem)
1165     THROWF(arg_error, EINVAL,
1166            "Cannot get the value of the NULL semaphore");
1167
1168   EnterCriticalSection(&(sem->value_lock));
1169   *svalue = sem->value;
1170   LeaveCriticalSection(&(sem->value_lock));
1171 }
1172
1173
1174 #endif
1175
1176 void xbt_os_thread_set_extra_data(void *data)
1177 {
1178   xbt_os_thread_self()->extra_data = data;
1179 }
1180
1181 void *xbt_os_thread_get_extra_data(void)
1182 {
1183   return xbt_os_thread_self()->extra_data;
1184 }