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