src/external/thread_mpi/src/pthreads.c

   1 /*
   2    This source code file is part of thread_mpi.
   3    Written by Sander Pronk, Erik Lindahl, and possibly others.
   4
   5    Copyright (c) 2009, Sander Pronk, Erik Lindahl.
   6    All rights reserved.
   7
   8    Redistribution and use in source and binary forms, with or without
   9    modification, are permitted provided that the following conditions are met:
  10    1) Redistributions of source code must retain the above copyright
  11    notice, this list of conditions and the following disclaimer.
  12    2) Redistributions in binary form must reproduce the above copyright
  13    notice, this list of conditions and the following disclaimer in the
  14    documentation and/or other materials provided with the distribution.
  15    3) Neither the name of the copyright holders nor the
  16    names of its contributors may be used to endorse or promote products
  17    derived from this software without specific prior written permission.
  18
  19    THIS SOFTWARE IS PROVIDED BY US ''AS IS'' AND ANY
  20    EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  21    WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  22    DISCLAIMED. IN NO EVENT SHALL WE BE LIABLE FOR ANY
  23    DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  24    (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  25    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  26    ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  27    (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  28    SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29
  30    If you want to redistribute modifications, please consider that
  31    scientific software is very special. Version control is crucial -
  32    bugs must be traceable. We will be happy to consider code for
  33    inclusion in the official distribution, but derived work should not
  34    be called official thread_mpi. Details are found in the README & COPYING
  35    files.
  36  */
  37
  38
  39
  40 /* Include the defines that determine which thread library to use.
  41  * We do not use HAVE_PTHREAD_H directly, since we might want to
  42  * turn off thread support explicity (e.g. for debugging).
  43  */
  44 #ifdef HAVE_TMPI_CONFIG_H
  45 #include "tmpi_config.h"
  46 #endif
  47
  48 #ifdef HAVE_CONFIG_H
  49 #include "config.h"
  50 #endif
  51
  52
  53 #ifdef THREAD_PTHREADS
  54
  55 #ifdef HAVE_PTHREAD_SETAFFINITY
  56 #define _GNU_SOURCE
  57 #endif
  58
  59 /* pthread.h must be the first header, apart from the defines in config.h */
  60 #include <pthread.h>
  61
  62
  63 #ifdef HAVE_UNISTD_H
  64 #include <unistd.h>
  65 #endif
  66
  67 #include <errno.h>
  68 #include <stdlib.h>
  69 #include <stdio.h>
  70 #include <stdarg.h>
  71
  72 #include "thread_mpi/atomic.h"
  73 #include "thread_mpi/threads.h"
  74 #include "impl.h"
  75 #include "unused.h"
  76
  77 #include "pthreads.h"
  78
  79 /* mutex for initializing mutexes */
  80 static pthread_mutex_t mutex_init = PTHREAD_MUTEX_INITIALIZER;
  81 /* mutex for initializing barriers */
  82 static pthread_mutex_t once_init = PTHREAD_MUTEX_INITIALIZER;
  83 /* mutex for initializing thread_conds */
  84 static pthread_mutex_t cond_init = PTHREAD_MUTEX_INITIALIZER;
  85 /* mutex for initializing barriers */
  86 static pthread_mutex_t barrier_init = PTHREAD_MUTEX_INITIALIZER;
  87
  88 /* mutex for managing  thread IDs */
  89 static pthread_mutex_t thread_id_mutex = PTHREAD_MUTEX_INITIALIZER;
  90 static pthread_key_t   thread_id_key;
  91 static int             thread_id_key_initialized = 0;
  92
  93
  94
  95
  96 enum tMPI_Thread_support tMPI_Thread_support(void)
  97 {
  98     return TMPI_THREAD_SUPPORT_YES;
  99 }
 100
 101
 102 int tMPI_Thread_get_hw_number(void)
 103 {
 104     int ret = 0;
 105 #ifdef HAVE_SYSCONF
 106 #if defined(_SC_NPROCESSORS_ONLN)
 107     ret = sysconf(_SC_NPROCESSORS_ONLN);
 108 #elif defined(_SC_NPROC_ONLN)
 109     ret = sysconf(_SC_NPROC_ONLN);
 110 #elif defined(_SC_NPROCESSORS_CONF)
 111     ret = sysconf(_SC_NPROCESSORS_CONF);
 112 #elif defined(_SC_NPROC_CONF)
 113     ret = sysconf(_SC_NPROC_CONF);
 114 #endif
 115 #endif
 116
 117     return ret;
 118 }
 119
 120 /* destructor for thread ids */
 121 static void tMPI_Destroy_thread_id(void* thread_id)
 122 {
 123     struct tMPI_Thread *thread = (struct tMPI_Thread*)thread_id;
 124     if (!thread->started_by_tmpi)
 125     {
 126         /* if the thread is started by tMPI, it must be freed in the join()
 127            call. */
 128         free(thread_id);
 129     }
 130 }
 131
 132
 133 /* Set the thread id var for this thread
 134     Returns a pointer to the thread object if succesful, NULL if ENOMEM */
 135 static struct tMPI_Thread* tMPI_Set_thread_id_key(int started_by_tmpi)
 136 {
 137     struct tMPI_Thread *th;
 138
 139     th = (struct tMPI_Thread*)malloc(sizeof(struct tMPI_Thread)*1);
 140     if (th == NULL)
 141     {
 142         return NULL;
 143     }
 144     th->th              = pthread_self();
 145     th->started_by_tmpi = started_by_tmpi;
 146     /* we ignore errors because any thread that needs this value will
 147        re-generate it in the next iteration. */
 148     pthread_setspecific(thread_id_key, th);
 149     return th;
 150 }
 151
 152 /* initialize the thread id vars if not already initialized */
 153 static int tMPI_Init_thread_ids(void)
 154 {
 155     int ret;
 156     ret = pthread_mutex_lock( &thread_id_mutex );
 157     if (ret != 0)
 158     {
 159         return ret;
 160     }
 161
 162     if (!thread_id_key_initialized)
 163     {
 164         /* initialize and set the thread id thread-specific variable */
 165         struct tMPI_Thread *th;
 166
 167         thread_id_key_initialized = 1;
 168         ret = pthread_key_create(&thread_id_key, tMPI_Destroy_thread_id);
 169         if (ret != 0)
 170         {
 171             goto err;
 172         }
 173         th = tMPI_Set_thread_id_key(0);
 174         if (th == NULL)
 175         {
 176             ret = ENOMEM;
 177             goto err;
 178         }
 179     }
 180
 181     ret = pthread_mutex_unlock( &thread_id_mutex );
 182     return ret;
 183 err:
 184     pthread_mutex_unlock( &thread_id_mutex );
 185     return ret;
 186 }
 187
 188 /* structure to hold the arguments for the thread_starter function */
 189 struct tMPI_Thread_starter
 190 {
 191     struct tMPI_Thread *thread;
 192     void               *(*start_routine)(void*);
 193     void               *arg;
 194     pthread_mutex_t     cond_lock; /* lock for initialization of thread
 195                                       structure */
 196 };
 197
 198 /* the thread_starter function that sets the thread id */
 199 #ifdef __MINGW32__
 200 __attribute__((force_align_arg_pointer))
 201 #endif
 202 static void *tMPI_Thread_starter(void *arg)
 203 {
 204     struct tMPI_Thread_starter *starter = (struct tMPI_Thread_starter *)arg;
 205     void *(*start_routine)(void*);
 206     void *parg;
 207     int   ret;
 208
 209     /* first wait for the parent thread to signal that the starter->thread
 210        structure is ready. That's done by unlocking the starter->cond_lock */
 211     ret = pthread_mutex_lock(&(starter->cond_lock));
 212     if (ret != 0)
 213     {
 214         return NULL;
 215     }
 216     ret = pthread_mutex_unlock(&(starter->cond_lock));
 217     if (ret != 0)
 218     {
 219         return NULL;
 220     }
 221
 222     /* now remember the tMPI_thread_t structure for this thread */
 223     ret = pthread_setspecific(thread_id_key, starter->thread);
 224     if (ret != 0)
 225     {
 226         return NULL;
 227     }
 228     start_routine = starter->start_routine;
 229     parg          = starter->arg;
 230
 231     /* deallocate the starter structure. Errors here are non-fatal. */
 232     pthread_mutex_destroy(&(starter->cond_lock));
 233     free(starter);
 234     return (*start_routine)(parg);
 235 }
 236
 237 int tMPI_Thread_create(tMPI_Thread_t *thread, void *(*start_routine)(void *),
 238                        void *arg)
 239 {
 240     int ret;
 241     struct tMPI_Thread_starter *starter;
 242
 243     if (thread == NULL)
 244     {
 245         return EINVAL;
 246     }
 247     tMPI_Init_thread_ids();
 248
 249     *thread = (struct tMPI_Thread*)malloc(sizeof(struct tMPI_Thread)*1);
 250     if (*thread == NULL)
 251     {
 252         return ENOMEM;
 253     }
 254     (*thread)->started_by_tmpi = 1;
 255     starter                    = (struct tMPI_Thread_starter*)
 256         malloc(sizeof(struct tMPI_Thread_starter)*1);
 257     if (starter == NULL)
 258     {
 259         return ENOMEM;
 260     }
 261     /* fill the starter structure */
 262     starter->thread        = *thread;
 263     starter->start_routine = start_routine;
 264     starter->arg           = arg;
 265
 266     ret = pthread_mutex_init(&(starter->cond_lock), NULL);
 267     if (ret != 0)
 268     {
 269         return ret;
 270     }
 271     /* now lock the mutex so we can unlock it once we know the data in
 272        thread->th is safe. */
 273     ret = pthread_mutex_lock(&(starter->cond_lock));
 274     if (ret != 0)
 275     {
 276         return ret;
 277     }
 278
 279     ret = pthread_create(&((*thread)->th), NULL, tMPI_Thread_starter,
 280                          (void*)starter);
 281     if (ret != 0)
 282     {
 283         return ret;
 284     }
 285
 286     /* Here we know thread->th is safe. */
 287     ret = pthread_mutex_unlock(&(starter->cond_lock));
 288
 289     return ret;
 290 }
 291
 292
 293
 294 int tMPI_Thread_join(tMPI_Thread_t thread, void **value_ptr)
 295 {
 296     int       ret;
 297     pthread_t th = thread->th;
 298
 299     ret = pthread_join( th, value_ptr );
 300     if (ret != 0)
 301     {
 302         return ret;
 303     }
 304     free(thread);
 305     return 0;
 306 }
 307
 308
 309 tMPI_Thread_t tMPI_Thread_self(void)
 310 {
 311     tMPI_Thread_t th;
 312     int           ret;
 313
 314     /* make sure the key var is set */
 315     ret = tMPI_Init_thread_ids();
 316     if (ret != 0)
 317     {
 318         return NULL;
 319     }
 320
 321     th = pthread_getspecific(thread_id_key);
 322
 323     /* check if it is already in our list */
 324     if (th == NULL)
 325     {
 326         th = tMPI_Set_thread_id_key(0);
 327     }
 328     return th;
 329 }
 330
 331 int tMPI_Thread_equal(tMPI_Thread_t t1, tMPI_Thread_t t2)
 332 {
 333     return pthread_equal(t1->th, t2->th);
 334 }
 335
 336
 337 enum tMPI_Thread_setaffinity_support tMPI_Thread_setaffinity_support(void)
 338 {
 339 #ifdef HAVE_PTHREAD_SETAFFINITY
 340     cpu_set_t set;
 341     int       ret;
 342
 343     /* run getaffinity to check whether we get back ENOSYS */
 344     ret = pthread_getaffinity_np(pthread_self(), sizeof(set), &set);
 345     if (ret == 0)
 346     {
 347         return TMPI_SETAFFINITY_SUPPORT_YES;
 348     }
 349     else
 350     {
 351         return TMPI_SETAFFINITY_SUPPORT_NO;
 352     }
 353 #else
 354     return TMPI_SETAFFINITY_SUPPORT_NO;
 355 #endif
 356 }
 357
 358
 359 /* set thread's own affinity to a processor number n */
 360 int tMPI_Thread_setaffinity_single(tMPI_Thread_t tmpi_unused thread,
 361                                    unsigned int  tmpi_unused nr)
 362 {
 363 #ifdef HAVE_PTHREAD_SETAFFINITY
 364     int       nt = tMPI_Thread_get_hw_number();
 365     cpu_set_t set;
 366
 367     if (nt < nr)
 368     {
 369         return TMPI_ERR_PROCNR;
 370     }
 371
 372     CPU_ZERO(&set);
 373     CPU_SET(nr, &set);
 374     return pthread_setaffinity_np(thread->th, sizeof(set), &set);
 375 #endif
 376     return 0;
 377 }
 378
 379
 380
 381
 382 int tMPI_Thread_mutex_init(tMPI_Thread_mutex_t *mtx)
 383 {
 384     int ret;
 385
 386     if (mtx == NULL)
 387     {
 388         return EINVAL;
 389     }
 390
 391     mtx->mutex = (struct tMPI_Mutex*)malloc(sizeof(struct tMPI_Mutex)*1);
 392     if (mtx->mutex == NULL)
 393     {
 394         return ENOMEM;
 395     }
 396     ret = pthread_mutex_init(&(mtx->mutex->mtx), NULL);
 397     if (ret != 0)
 398     {
 399         return ret;
 400     }
 401
 402 #ifndef TMPI_NO_ATOMICS
 403     tMPI_Atomic_set(&(mtx->initialized), 1);
 404 #else
 405     mtx->initialized.value = 1;
 406 #endif
 407     return 0;
 408 }
 409
 410 static inline int tMPI_Thread_mutex_init_once(tMPI_Thread_mutex_t *mtx)
 411 {
 412     int ret = 0;
 413
 414 #ifndef TMPI_NO_ATOMICS
 415     /* check whether the mutex is initialized */
 416     if (tMPI_Atomic_get( &(mtx->initialized)  ) == 0)
 417 #endif
 418     {
 419         /* we're relying on the memory barrier semantics of mutex_lock/unlock
 420            for the check preceding this function call to have worked */
 421         ret = pthread_mutex_lock( &(mutex_init) );
 422         if (ret != 0)
 423         {
 424             return ret;
 425         }
 426
 427         if (mtx->mutex == NULL)
 428         {
 429             mtx->mutex = (struct tMPI_Mutex*)malloc(sizeof(struct tMPI_Mutex));
 430             if (mtx->mutex == NULL)
 431             {
 432                 ret = ENOMEM;
 433                 goto err;
 434             }
 435             ret = pthread_mutex_init( &(mtx->mutex->mtx), NULL);
 436             if (ret != 0)
 437             {
 438                 goto err;
 439             }
 440         }
 441         ret = pthread_mutex_unlock( &(mutex_init) );
 442     }
 443     return ret;
 444 err:
 445     pthread_mutex_unlock( &(mutex_init) );
 446     return ret;
 447 }
 448
 449
 450 int tMPI_Thread_mutex_destroy(tMPI_Thread_mutex_t *mtx)
 451 {
 452     int ret;
 453
 454     if (mtx == NULL)
 455     {
 456         return EINVAL;
 457     }
 458
 459     ret = pthread_mutex_destroy( &(mtx->mutex->mtx) );
 460     if (ret != 0)
 461     {
 462         return ret;
 463     }
 464     free(mtx->mutex);
 465     return ret;
 466 }
 467
 468
 469
 470 int tMPI_Thread_mutex_lock(tMPI_Thread_mutex_t *mtx)
 471 {
 472     int ret;
 473
 474     /* check whether the mutex is initialized */
 475     ret = tMPI_Thread_mutex_init_once(mtx);
 476     if (ret != 0)
 477     {
 478         return ret;
 479     }
 480
 481     ret = pthread_mutex_lock(&(mtx->mutex->mtx));
 482     return ret;
 483 }
 484
 485
 486
 487
 488 int tMPI_Thread_mutex_trylock(tMPI_Thread_mutex_t *mtx)
 489 {
 490     int ret;
 491
 492     /* check whether the mutex is initialized */
 493     ret = tMPI_Thread_mutex_init_once(mtx);
 494     if (ret != 0)
 495     {
 496         return ret;
 497     }
 498
 499     ret = pthread_mutex_trylock(&(mtx->mutex->mtx));
 500     return ret;
 501 }
 502
 503
 504
 505 int tMPI_Thread_mutex_unlock(tMPI_Thread_mutex_t *mtx)
 506 {
 507     int ret;
 508
 509     /* check whether the mutex is initialized */
 510     ret = tMPI_Thread_mutex_init_once(mtx);
 511     if (ret != 0)
 512     {
 513         return ret;
 514     }
 515
 516     ret = pthread_mutex_unlock(&(mtx->mutex->mtx));
 517     return ret;
 518 }
 519
 520
 521
 522 int tMPI_Thread_key_create(tMPI_Thread_key_t *key, void (*destructor)(void *))
 523 {
 524     int ret;
 525
 526     if (key == NULL)
 527     {
 528         return EINVAL;
 529     }
 530
 531
 532     key->key = (struct tMPI_Thread_key*)malloc(sizeof(struct
 533                                                       tMPI_Thread_key)*1);
 534     if (key->key == NULL)
 535     {
 536         return ENOMEM;
 537     }
 538     ret = pthread_key_create(&((key)->key->pkey), destructor);
 539     if (ret != 0)
 540     {
 541         return ret;
 542     }
 543
 544     tMPI_Atomic_set(&(key->initialized), 1);
 545     return 0;
 546 }
 547
 548
 549 int tMPI_Thread_key_delete(tMPI_Thread_key_t key)
 550 {
 551     int ret;
 552
 553     ret = pthread_key_delete((key.key->pkey));
 554     if (ret != 0)
 555     {
 556         return ret;
 557     }
 558     free(key.key);
 559
 560     return 0;
 561 }
 562
 563
 564
 565 void * tMPI_Thread_getspecific(tMPI_Thread_key_t key)
 566 {
 567     void *p = NULL;
 568
 569     p = pthread_getspecific((key.key->pkey));
 570
 571     return p;
 572 }
 573
 574
 575 int tMPI_Thread_setspecific(tMPI_Thread_key_t key, void *value)
 576 {
 577     int ret;
 578
 579     ret = pthread_setspecific((key.key->pkey), value);
 580
 581     return ret;
 582 }
 583
 584
 585 int tMPI_Thread_once(tMPI_Thread_once_t *once_control,
 586                      void                (*init_routine)(void))
 587 {
 588     int ret;
 589     if (!once_control || !init_routine)
 590     {
 591         return EINVAL;
 592     }
 593
 594     /* really ugly hack - and it's slow... */
 595     ret = pthread_mutex_lock( &once_init );
 596     if (ret != 0)
 597     {
 598         return ret;
 599     }
 600     if (tMPI_Atomic_get(&(once_control->once)) == 0)
 601     {
 602         (*init_routine)();
 603         tMPI_Atomic_set(&(once_control->once), 1);
 604     }
 605     ret = pthread_mutex_unlock( &once_init );
 606
 607     return ret;
 608 }
 609
 610
 611
 612
 613 int tMPI_Thread_cond_init(tMPI_Thread_cond_t *cond)
 614 {
 615     int ret;
 616
 617     if (cond == NULL)
 618     {
 619         return EINVAL;
 620     }
 621
 622     cond->condp = (struct tMPI_Thread_cond*)malloc(
 623                 sizeof(struct tMPI_Thread_cond));
 624     if (cond->condp == NULL)
 625     {
 626         return ENOMEM;
 627     }
 628
 629     ret = pthread_cond_init(&(cond->condp->cond), NULL);
 630     if (ret != 0)
 631     {
 632         return ret;
 633     }
 634     tMPI_Atomic_set(&(cond->initialized), 1);
 635     tMPI_Atomic_memory_barrier();
 636
 637     return 0;
 638 }
 639
 640
 641 static int tMPI_Thread_cond_init_once(tMPI_Thread_cond_t *cond)
 642 {
 643     int ret = 0;
 644
 645     /* we're relying on the memory barrier semantics of mutex_lock/unlock
 646        for the check preceding this function call to have worked */
 647     ret = pthread_mutex_lock( &(cond_init) );
 648     if (ret != 0)
 649     {
 650         return ret;
 651     }
 652     if (cond->condp == NULL)
 653     {
 654         cond->condp = (struct tMPI_Thread_cond*)
 655             malloc(sizeof(struct tMPI_Thread_cond)*1);
 656         if (cond->condp == NULL)
 657         {
 658             ret = ENOMEM;
 659             goto err;
 660         }
 661         ret = pthread_cond_init( &(cond->condp->cond), NULL);
 662         if (ret != 0)
 663         {
 664             goto err;
 665         }
 666     }
 667     ret = pthread_mutex_unlock( &(cond_init) );
 668     return ret;
 669 err:
 670     /* try to unlock anyway */
 671     pthread_mutex_unlock( &(cond_init) );
 672     return ret;
 673 }
 674
 675
 676
 677 int tMPI_Thread_cond_destroy(tMPI_Thread_cond_t *cond)
 678 {
 679     int ret;
 680
 681     if (cond == NULL)
 682     {
 683         return EINVAL;
 684     }
 685
 686     ret = pthread_cond_destroy(&(cond->condp->cond));
 687     if (ret != 0)
 688     {
 689         return ret;
 690     }
 691     free(cond->condp);
 692
 693     return 0;
 694 }
 695
 696
 697 int tMPI_Thread_cond_wait(tMPI_Thread_cond_t *cond, tMPI_Thread_mutex_t *mtx)
 698 {
 699     int ret;
 700
 701     /* check whether the condition is initialized */
 702     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
 703     {
 704         ret = tMPI_Thread_cond_init_once(cond);
 705         if (ret != 0)
 706         {
 707             return ret;
 708         }
 709     }
 710     /* the mutex must have been initialized because it should be locked here */
 711
 712     ret = pthread_cond_wait( &(cond->condp->cond), &(mtx->mutex->mtx) );
 713
 714     return ret;
 715 }
 716
 717
 718
 719
 720 int tMPI_Thread_cond_signal(tMPI_Thread_cond_t *cond)
 721 {
 722     int ret;
 723
 724     /* check whether the condition is initialized */
 725     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
 726     {
 727         ret = tMPI_Thread_cond_init_once(cond);
 728         if (ret != 0)
 729         {
 730             return ret;
 731         }
 732     }
 733
 734     ret = pthread_cond_signal( &(cond->condp->cond) );
 735
 736     return ret;
 737 }
 738
 739
 740
 741 int tMPI_Thread_cond_broadcast(tMPI_Thread_cond_t *cond)
 742 {
 743     int ret;
 744
 745     /* check whether the condition is initialized */
 746     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
 747     {
 748         ret = tMPI_Thread_cond_init_once(cond);
 749         if (ret != 0)
 750         {
 751             return ret;
 752         }
 753     }
 754
 755     ret = pthread_cond_broadcast( &(cond->condp->cond) );
 756
 757     return ret;
 758 }
 759
 760
 761
 762
 763 void tMPI_Thread_exit(void *value_ptr)
 764 {
 765     pthread_exit(value_ptr);
 766 }
 767
 768
 769 int tMPI_Thread_cancel(tMPI_Thread_t thread)
 770 {
 771     #ifdef __native_client__
 772     return ENOSYS;
 773     #endif
 774     return pthread_cancel(thread->th);
 775 }
 776
 777
 778
 779
 780 int tMPI_Thread_barrier_init(tMPI_Thread_barrier_t *barrier, int n)
 781 {
 782     int ret;
 783     /*tMPI_Thread_pthread_barrier_t *p;*/
 784
 785     if (barrier == NULL)
 786     {
 787         return EINVAL;
 788     }
 789
 790     barrier->barrierp = (struct tMPI_Thread_barrier*)
 791         malloc(sizeof(struct tMPI_Thread_barrier)*1);
 792     if (barrier->barrierp == NULL)
 793     {
 794         return ENOMEM;
 795     }
 796
 797     ret = pthread_mutex_init(&(barrier->barrierp->mutex), NULL);
 798     if (ret != 0)
 799     {
 800         return ret;
 801     }
 802
 803     ret = pthread_cond_init(&(barrier->barrierp->cv), NULL);
 804     if (ret != 0)
 805     {
 806         return ret;
 807     }
 808
 809     barrier->threshold = n;
 810     barrier->count     = n;
 811     barrier->cycle     = 0;
 812
 813     tMPI_Atomic_set(&(barrier->initialized), 1);
 814     return 0;
 815 }
 816
 817 static int tMPI_Thread_barrier_init_once(tMPI_Thread_barrier_t *barrier)
 818 {
 819     int ret = 0;
 820
 821     /* we're relying on the memory barrier semantics of mutex_lock/unlock
 822        for the check preceding this function call to have worked */
 823     ret = pthread_mutex_lock( &(barrier_init) );
 824     if (ret != 0)
 825     {
 826         return ret;
 827     }
 828
 829     if (barrier->barrierp == NULL)
 830     {
 831         barrier->barrierp = (struct tMPI_Thread_barrier*)
 832             malloc(sizeof(struct tMPI_Thread_barrier)*1);
 833         if (barrier->barrierp == NULL)
 834         {
 835             ret = ENOMEM;
 836             goto err;
 837         }
 838
 839         ret = pthread_mutex_init(&(barrier->barrierp->mutex), NULL);
 840
 841         if (ret != 0)
 842         {
 843             goto err;
 844         }
 845
 846         ret = pthread_cond_init(&(barrier->barrierp->cv), NULL);
 847
 848         if (ret != 0)
 849         {
 850             goto err;
 851         }
 852     }
 853     ret = pthread_mutex_unlock( &(barrier_init) );
 854     return ret;
 855 err:
 856     pthread_mutex_unlock( &(barrier_init) );
 857     return ret;
 858 }
 859
 860
 861
 862
 863 int tMPI_Thread_barrier_destroy(tMPI_Thread_barrier_t *barrier)
 864 {
 865     int ret;
 866
 867     if (barrier == NULL)
 868     {
 869         return EINVAL;
 870     }
 871
 872     ret = pthread_mutex_destroy(&(barrier->barrierp->mutex));
 873     if (ret != 0)
 874     {
 875         return ret;
 876     }
 877     ret = pthread_cond_destroy(&(barrier->barrierp->cv));
 878     if (ret != 0)
 879     {
 880         return ret;
 881     }
 882
 883     free(barrier->barrierp);
 884
 885     return 0;
 886 }
 887
 888
 889 int tMPI_Thread_barrier_wait(tMPI_Thread_barrier_t * barrier)
 890 {
 891     int cycle;
 892     int ret;
 893
 894     /* check whether the barrier is initialized */
 895     if (tMPI_Atomic_get( &(barrier->initialized)  ) == 0)
 896     {
 897         tMPI_Thread_barrier_init_once(barrier);
 898     }
 899
 900
 901     ret = pthread_mutex_lock(&barrier->barrierp->mutex);
 902     if (ret != 0)
 903     {
 904         return ret;
 905     }
 906
 907     cycle = barrier->cycle;
 908
 909     /* Decrement the count atomically and check if it is zero.
 910      * This will only be true for the last thread calling us.
 911      */
 912     if (--barrier->count <= 0)
 913     {
 914         barrier->cycle = !barrier->cycle;
 915         barrier->count = barrier->threshold;
 916         ret            = pthread_cond_broadcast(&barrier->barrierp->cv);
 917
 918         if (ret == 0)
 919         {
 920             goto err;
 921         }
 922     }
 923     else
 924     {
 925         while (cycle == barrier->cycle)
 926         {
 927             ret = pthread_cond_wait(&barrier->barrierp->cv,
 928                                     &barrier->barrierp->mutex);
 929             if (ret != 0)
 930             {
 931                 goto err;
 932             }
 933         }
 934     }
 935
 936     ret = pthread_mutex_unlock(&barrier->barrierp->mutex);
 937     return ret;
 938 err:
 939     pthread_mutex_unlock(&barrier->barrierp->mutex);
 940     return ret;
 941
 942 }
 943
 944 #else
 945
 946 /* just to have some symbols */
 947 int tMPI_Thread_pthreads = 0;
 948
 949 #endif /* THREAD_PTHREADS */