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 #else
 376     return 0;
 377 #endif
 378 }
 379
 380
 381
 382
 383 int tMPI_Thread_mutex_init(tMPI_Thread_mutex_t *mtx)
 384 {
 385     int ret;
 386
 387     if (mtx == NULL)
 388     {
 389         return EINVAL;
 390     }
 391
 392     mtx->mutex = (struct tMPI_Mutex*)malloc(sizeof(struct tMPI_Mutex)*1);
 393     if (mtx->mutex == NULL)
 394     {
 395         return ENOMEM;
 396     }
 397     ret = pthread_mutex_init(&(mtx->mutex->mtx), NULL);
 398     if (ret != 0)
 399     {
 400         return ret;
 401     }
 402
 403 #ifndef TMPI_NO_ATOMICS
 404     tMPI_Atomic_set(&(mtx->initialized), 1);
 405 #else
 406     mtx->initialized.value = 1;
 407 #endif
 408     return 0;
 409 }
 410
 411 static inline int tMPI_Thread_mutex_init_once(tMPI_Thread_mutex_t *mtx)
 412 {
 413     int ret = 0;
 414
 415 #ifndef TMPI_NO_ATOMICS
 416     /* check whether the mutex is initialized */
 417     if (tMPI_Atomic_get( &(mtx->initialized)  ) == 0)
 418 #endif
 419     {
 420         /* we're relying on the memory barrier semantics of mutex_lock/unlock
 421            for the check preceding this function call to have worked */
 422         ret = pthread_mutex_lock( &(mutex_init) );
 423         if (ret != 0)
 424         {
 425             return ret;
 426         }
 427
 428         if (mtx->mutex == NULL)
 429         {
 430             mtx->mutex = (struct tMPI_Mutex*)malloc(sizeof(struct tMPI_Mutex));
 431             if (mtx->mutex == NULL)
 432             {
 433                 ret = ENOMEM;
 434                 goto err;
 435             }
 436             ret = pthread_mutex_init( &(mtx->mutex->mtx), NULL);
 437             if (ret != 0)
 438             {
 439                 goto err;
 440             }
 441         }
 442         ret = pthread_mutex_unlock( &(mutex_init) );
 443     }
 444     return ret;
 445 err:
 446     pthread_mutex_unlock( &(mutex_init) );
 447     return ret;
 448 }
 449
 450
 451 int tMPI_Thread_mutex_destroy(tMPI_Thread_mutex_t *mtx)
 452 {
 453     int ret;
 454
 455     if (mtx == NULL)
 456     {
 457         return EINVAL;
 458     }
 459
 460     ret = pthread_mutex_destroy( &(mtx->mutex->mtx) );
 461     if (ret != 0)
 462     {
 463         return ret;
 464     }
 465     free(mtx->mutex);
 466     return ret;
 467 }
 468
 469
 470
 471 int tMPI_Thread_mutex_lock(tMPI_Thread_mutex_t *mtx)
 472 {
 473     int ret;
 474
 475     /* check whether the mutex is initialized */
 476     ret = tMPI_Thread_mutex_init_once(mtx);
 477     if (ret != 0)
 478     {
 479         return ret;
 480     }
 481
 482     ret = pthread_mutex_lock(&(mtx->mutex->mtx));
 483     return ret;
 484 }
 485
 486
 487
 488
 489 int tMPI_Thread_mutex_trylock(tMPI_Thread_mutex_t *mtx)
 490 {
 491     int ret;
 492
 493     /* check whether the mutex is initialized */
 494     ret = tMPI_Thread_mutex_init_once(mtx);
 495     if (ret != 0)
 496     {
 497         return ret;
 498     }
 499
 500     ret = pthread_mutex_trylock(&(mtx->mutex->mtx));
 501     return ret;
 502 }
 503
 504
 505
 506 int tMPI_Thread_mutex_unlock(tMPI_Thread_mutex_t *mtx)
 507 {
 508     int ret;
 509
 510     /* check whether the mutex is initialized */
 511     ret = tMPI_Thread_mutex_init_once(mtx);
 512     if (ret != 0)
 513     {
 514         return ret;
 515     }
 516
 517     ret = pthread_mutex_unlock(&(mtx->mutex->mtx));
 518     return ret;
 519 }
 520
 521
 522
 523 int tMPI_Thread_key_create(tMPI_Thread_key_t *key, void (*destructor)(void *))
 524 {
 525     int ret;
 526
 527     if (key == NULL)
 528     {
 529         return EINVAL;
 530     }
 531
 532
 533     key->key = (struct tMPI_Thread_key*)malloc(sizeof(struct
 534                                                       tMPI_Thread_key)*1);
 535     if (key->key == NULL)
 536     {
 537         return ENOMEM;
 538     }
 539     ret = pthread_key_create(&((key)->key->pkey), destructor);
 540     if (ret != 0)
 541     {
 542         return ret;
 543     }
 544
 545     tMPI_Atomic_set(&(key->initialized), 1);
 546     return 0;
 547 }
 548
 549
 550 int tMPI_Thread_key_delete(tMPI_Thread_key_t key)
 551 {
 552     int ret;
 553
 554     ret = pthread_key_delete((key.key->pkey));
 555     if (ret != 0)
 556     {
 557         return ret;
 558     }
 559     free(key.key);
 560
 561     return 0;
 562 }
 563
 564
 565
 566 void * tMPI_Thread_getspecific(tMPI_Thread_key_t key)
 567 {
 568     void *p = NULL;
 569
 570     p = pthread_getspecific((key.key->pkey));
 571
 572     return p;
 573 }
 574
 575
 576 int tMPI_Thread_setspecific(tMPI_Thread_key_t key, void *value)
 577 {
 578     int ret;
 579
 580     ret = pthread_setspecific((key.key->pkey), value);
 581
 582     return ret;
 583 }
 584
 585
 586 int tMPI_Thread_once(tMPI_Thread_once_t *once_control,
 587                      void                (*init_routine)(void))
 588 {
 589     int ret;
 590     if (!once_control || !init_routine)
 591     {
 592         return EINVAL;
 593     }
 594
 595     /* really ugly hack - and it's slow... */
 596     ret = pthread_mutex_lock( &once_init );
 597     if (ret != 0)
 598     {
 599         return ret;
 600     }
 601     if (tMPI_Atomic_get(&(once_control->once)) == 0)
 602     {
 603         (*init_routine)();
 604         tMPI_Atomic_set(&(once_control->once), 1);
 605     }
 606     ret = pthread_mutex_unlock( &once_init );
 607
 608     return ret;
 609 }
 610
 611
 612
 613
 614 int tMPI_Thread_cond_init(tMPI_Thread_cond_t *cond)
 615 {
 616     int ret;
 617
 618     if (cond == NULL)
 619     {
 620         return EINVAL;
 621     }
 622
 623     cond->condp = (struct tMPI_Thread_cond*)malloc(
 624                 sizeof(struct tMPI_Thread_cond));
 625     if (cond->condp == NULL)
 626     {
 627         return ENOMEM;
 628     }
 629
 630     ret = pthread_cond_init(&(cond->condp->cond), NULL);
 631     if (ret != 0)
 632     {
 633         return ret;
 634     }
 635     tMPI_Atomic_set(&(cond->initialized), 1);
 636     tMPI_Atomic_memory_barrier();
 637
 638     return 0;
 639 }
 640
 641
 642 static int tMPI_Thread_cond_init_once(tMPI_Thread_cond_t *cond)
 643 {
 644     int ret = 0;
 645
 646     /* we're relying on the memory barrier semantics of mutex_lock/unlock
 647        for the check preceding this function call to have worked */
 648     ret = pthread_mutex_lock( &(cond_init) );
 649     if (ret != 0)
 650     {
 651         return ret;
 652     }
 653     if (cond->condp == NULL)
 654     {
 655         cond->condp = (struct tMPI_Thread_cond*)
 656             malloc(sizeof(struct tMPI_Thread_cond)*1);
 657         if (cond->condp == NULL)
 658         {
 659             ret = ENOMEM;
 660             goto err;
 661         }
 662         ret = pthread_cond_init( &(cond->condp->cond), NULL);
 663         if (ret != 0)
 664         {
 665             goto err;
 666         }
 667     }
 668     ret = pthread_mutex_unlock( &(cond_init) );
 669     return ret;
 670 err:
 671     /* try to unlock anyway */
 672     pthread_mutex_unlock( &(cond_init) );
 673     return ret;
 674 }
 675
 676
 677
 678 int tMPI_Thread_cond_destroy(tMPI_Thread_cond_t *cond)
 679 {
 680     int ret;
 681
 682     if (cond == NULL)
 683     {
 684         return EINVAL;
 685     }
 686
 687     ret = pthread_cond_destroy(&(cond->condp->cond));
 688     if (ret != 0)
 689     {
 690         return ret;
 691     }
 692     free(cond->condp);
 693
 694     return 0;
 695 }
 696
 697
 698 int tMPI_Thread_cond_wait(tMPI_Thread_cond_t *cond, tMPI_Thread_mutex_t *mtx)
 699 {
 700     int ret;
 701
 702     /* check whether the condition is initialized */
 703     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
 704     {
 705         ret = tMPI_Thread_cond_init_once(cond);
 706         if (ret != 0)
 707         {
 708             return ret;
 709         }
 710     }
 711     /* the mutex must have been initialized because it should be locked here */
 712
 713     ret = pthread_cond_wait( &(cond->condp->cond), &(mtx->mutex->mtx) );
 714
 715     return ret;
 716 }
 717
 718
 719
 720
 721 int tMPI_Thread_cond_signal(tMPI_Thread_cond_t *cond)
 722 {
 723     int ret;
 724
 725     /* check whether the condition is initialized */
 726     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
 727     {
 728         ret = tMPI_Thread_cond_init_once(cond);
 729         if (ret != 0)
 730         {
 731             return ret;
 732         }
 733     }
 734
 735     ret = pthread_cond_signal( &(cond->condp->cond) );
 736
 737     return ret;
 738 }
 739
 740
 741
 742 int tMPI_Thread_cond_broadcast(tMPI_Thread_cond_t *cond)
 743 {
 744     int ret;
 745
 746     /* check whether the condition is initialized */
 747     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
 748     {
 749         ret = tMPI_Thread_cond_init_once(cond);
 750         if (ret != 0)
 751         {
 752             return ret;
 753         }
 754     }
 755
 756     ret = pthread_cond_broadcast( &(cond->condp->cond) );
 757
 758     return ret;
 759 }
 760
 761
 762
 763
 764 void tMPI_Thread_exit(void *value_ptr)
 765 {
 766     pthread_exit(value_ptr);
 767 }
 768
 769
 770 int tMPI_Thread_cancel(tMPI_Thread_t thread)
 771 {
 772     #ifdef __native_client__
 773     return ENOSYS;
 774     #endif
 775     return pthread_cancel(thread->th);
 776 }
 777
 778
 779
 780
 781 int tMPI_Thread_barrier_init(tMPI_Thread_barrier_t *barrier, int n)
 782 {
 783     int ret;
 784     /*tMPI_Thread_pthread_barrier_t *p;*/
 785
 786     if (barrier == NULL)
 787     {
 788         return EINVAL;
 789     }
 790
 791     barrier->barrierp = (struct tMPI_Thread_barrier*)
 792         malloc(sizeof(struct tMPI_Thread_barrier)*1);
 793     if (barrier->barrierp == NULL)
 794     {
 795         return ENOMEM;
 796     }
 797
 798     ret = pthread_mutex_init(&(barrier->barrierp->mutex), NULL);
 799     if (ret != 0)
 800     {
 801         return ret;
 802     }
 803
 804     ret = pthread_cond_init(&(barrier->barrierp->cv), NULL);
 805     if (ret != 0)
 806     {
 807         return ret;
 808     }
 809
 810     barrier->threshold = n;
 811     barrier->count     = n;
 812     barrier->cycle     = 0;
 813
 814     tMPI_Atomic_set(&(barrier->initialized), 1);
 815     return 0;
 816 }
 817
 818 static int tMPI_Thread_barrier_init_once(tMPI_Thread_barrier_t *barrier)
 819 {
 820     int ret = 0;
 821
 822     /* we're relying on the memory barrier semantics of mutex_lock/unlock
 823        for the check preceding this function call to have worked */
 824     ret = pthread_mutex_lock( &(barrier_init) );
 825     if (ret != 0)
 826     {
 827         return ret;
 828     }
 829
 830     if (barrier->barrierp == NULL)
 831     {
 832         barrier->barrierp = (struct tMPI_Thread_barrier*)
 833             malloc(sizeof(struct tMPI_Thread_barrier)*1);
 834         if (barrier->barrierp == NULL)
 835         {
 836             ret = ENOMEM;
 837             goto err;
 838         }
 839
 840         ret = pthread_mutex_init(&(barrier->barrierp->mutex), NULL);
 841
 842         if (ret != 0)
 843         {
 844             goto err;
 845         }
 846
 847         ret = pthread_cond_init(&(barrier->barrierp->cv), NULL);
 848
 849         if (ret != 0)
 850         {
 851             goto err;
 852         }
 853     }
 854     ret = pthread_mutex_unlock( &(barrier_init) );
 855     return ret;
 856 err:
 857     pthread_mutex_unlock( &(barrier_init) );
 858     return ret;
 859 }
 860
 861
 862
 863
 864 int tMPI_Thread_barrier_destroy(tMPI_Thread_barrier_t *barrier)
 865 {
 866     int ret;
 867
 868     if (barrier == NULL)
 869     {
 870         return EINVAL;
 871     }
 872
 873     ret = pthread_mutex_destroy(&(barrier->barrierp->mutex));
 874     if (ret != 0)
 875     {
 876         return ret;
 877     }
 878     ret = pthread_cond_destroy(&(barrier->barrierp->cv));
 879     if (ret != 0)
 880     {
 881         return ret;
 882     }
 883
 884     free(barrier->barrierp);
 885
 886     return 0;
 887 }
 888
 889
 890 int tMPI_Thread_barrier_wait(tMPI_Thread_barrier_t * barrier)
 891 {
 892     int cycle;
 893     int ret;
 894
 895     /* check whether the barrier is initialized */
 896     if (tMPI_Atomic_get( &(barrier->initialized)  ) == 0)
 897     {
 898         tMPI_Thread_barrier_init_once(barrier);
 899     }
 900
 901
 902     ret = pthread_mutex_lock(&barrier->barrierp->mutex);
 903     if (ret != 0)
 904     {
 905         return ret;
 906     }
 907
 908     cycle = barrier->cycle;
 909
 910     /* Decrement the count atomically and check if it is zero.
 911      * This will only be true for the last thread calling us.
 912      */
 913     if (--barrier->count <= 0)
 914     {
 915         barrier->cycle = !barrier->cycle;
 916         barrier->count = barrier->threshold;
 917         ret            = pthread_cond_broadcast(&barrier->barrierp->cv);
 918
 919         if (ret == 0)
 920         {
 921             goto err;
 922         }
 923     }
 924     else
 925     {
 926         while (cycle == barrier->cycle)
 927         {
 928             ret = pthread_cond_wait(&barrier->barrierp->cv,
 929                                     &barrier->barrierp->mutex);
 930             if (ret != 0)
 931             {
 932                 goto err;
 933             }
 934         }
 935     }
 936
 937     ret = pthread_mutex_unlock(&barrier->barrierp->mutex);
 938     return ret;
 939 err:
 940     pthread_mutex_unlock(&barrier->barrierp->mutex);
 941     return ret;
 942
 943 }
 944
 945 #else
 946
 947 /* just to have some symbols */
 948 int tMPI_Thread_pthreads = 0;
 949
 950 #endif /* THREAD_PTHREADS */