src/xbt/xbt_os_thread.c

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