src/gromacs/gmxlib/thread_mpi/winthreads.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
  45 #ifdef HAVE_TMPI_CONFIG_H
  46 #include "tmpi_config.h"
  47 #endif
  48
  49 #ifdef HAVE_CONFIG_H
  50 #include "config.h"
  51 #endif
  52
  53
  54 #ifdef THREAD_WINDOWS
  55
  56 /* the win32 header */
  57 #include <windows.h>
  58
  59
  60 #include <errno.h>
  61 #include <stdlib.h>
  62 #include <stdio.h>
  63 #include <stdarg.h>
  64
  65
  66 #include "thread_mpi/atomic.h"
  67 #include "thread_mpi/threads.h"
  68 #include "impl.h"
  69
  70 #include "winthreads.h"
  71
  72 /*! \brief System mutex for all one-time initialization
  73  *
  74  *  This static variable is necessary in order to make the header file
  75  *  independent of the thread library implementation. Anyway, it
  76  *  will only be locked a handful of times at the start of program execution.
  77  */
  78
  79 static CRITICAL_SECTION mutex_init;   /* mutex for initializing mutexes */
  80 static CRITICAL_SECTION once_init;    /* mutex for initializing barriers */
  81 static CRITICAL_SECTION cond_init;    /* mutex for initializing thread_conds */
  82 static CRITICAL_SECTION barrier_init; /* mutex for initializing barriers */
  83
  84
  85 /* spinlock for initializing the above mutexes */
  86 static tMPI_Spinlock_t init_init = TMPI_SPINLOCK_INITIALIZER;
  87
  88 /* whether tMPI_Thread_create has initialized these mutexes */
  89 static tMPI_Atomic_t init_inited = { 0 };
  90
  91 /* whether the main thread affinity has been set */
  92 static tMPI_Spinlock_t main_thread_aff_lock = TMPI_SPINLOCK_INITIALIZER;
  93 static tMPI_Atomic_t   main_thread_aff_set  = { 0 };
  94
  95 /* mutex for managing  thread IDs */
  96 static CRITICAL_SECTION thread_id_list_lock;
  97 typedef struct
  98 {
  99     DWORD               thread_id; /* the thread ID as returned by GetCurrentTreadID() */
 100     struct tMPI_Thread* th;        /* the associated tMPI thread structure */
 101 } thread_id_list_t;
 102 /* the size of the thrread id list */
 103 static int               Nalloc_thread_id_list = 0;
 104 /* the number of elements in the thread id list */
 105 static int               N_thread_id_list = 0;
 106 /* the thread ID list */
 107 static thread_id_list_t *thread_id_list;
 108
 109
 110
 111 /* data structure to keep track of thread key destructors. */
 112 typedef struct
 113 {
 114     void (*destructor)(void*);
 115     DWORD key;
 116 } thread_key_destructors;
 117
 118 static thread_key_destructors *destructors = NULL;
 119
 120
 121
 122 /*
 123     NUMA and Processor Group awareness support.
 124
 125     NUMA support is implemented to maximize the chance that memory access
 126     patterns remain Local to the NUMA node.
 127     NUMA node processor affinity is utilized to prevent scheduler associated
 128     drift across NUMA nodes.
 129     Process Group support is implemented to enable > 64 processors to be
 130     utilized.  This is only supported when building 64bit.
 131
 132     The high level approach is:
 133     1. Build a description of CPU topology, including processor numbers, NUMA
 134         node numbers, and affinity masks.
 135     2. For processor intensive worker threads, create threads such that
 136         the processor affinity and thread stack is kept local within a NUMA node.
 137     3. Employ simple round-robin affinity and node assignment approach when
 138         creating threads.
 139     4. Use GetProcAddress() to obtain function pointers to functions that
 140         are operating system version dependent, to allow maximum binary
 141         compatibility.
 142
 143     Scott Field (sfield@microsoft.com)      Jan-2011
 144  */
 145
 146
 147 typedef struct {
 148     PROCESSOR_NUMBER    ProcessorNumber;
 149     GROUP_AFFINITY      GroupAffinity;
 150     USHORT              NumaNodeNumber;
 151 } MPI_NUMA_PROCESSOR_INFO;
 152
 153
 154 /* thread/processor index, to allow setting round-robin affinity. */
 155 volatile ULONG           g_ulThreadIndex;
 156 /* a value of zero implies the system is not NUMA */
 157 ULONG                    g_ulHighestNumaNodeNumber = 0;
 158 /* total number of processors in g_MPI_ProcessInfo array */
 159 ULONG                    g_ulTotalProcessors;
 160 /* array describing available processors, affinity masks, and NUMA node */
 161 MPI_NUMA_PROCESSOR_INFO *g_MPI_ProcessorInfo = NULL;
 162
 163 /* function prototypes and variables to support obtaining function addresses
 164    dynamically -- supports down-level operating systems */
 165
 166 typedef BOOL (WINAPI *func_GetNumaHighestNodeNumber_t)( PULONG
 167                                                         HighestNodeNumber );
 168 typedef DWORD (WINAPI *func_SetThreadIdealProcessor_t)( HANDLE hThread,
 169                                                         DWORD dwIdealProcessor );
 170 typedef BOOL (WINAPI *func_SetThreadGroupAffinity_t)( HANDLE hThread,
 171                                                       const GROUP_AFFINITY *GroupAffinity,
 172                                                       PGROUP_AFFINITY PreviousGroupAffinity );
 173 typedef BOOL (WINAPI *func_SetThreadIdealProcessorEx_t)( HANDLE hThread,
 174                                                          PPROCESSOR_NUMBER lpIdealProcessor,
 175                                                          PPROCESSOR_NUMBER lpPreviousIdealProcessor );
 176 typedef BOOL (WINAPI *func_GetNumaNodeProcessorMaskEx_t)( USHORT Node,
 177                                                           PGROUP_AFFINITY ProcessorMask );
 178 typedef BOOL (WINAPI *func_GetNumaProcessorNodeEx_t)(
 179         PPROCESSOR_NUMBER Processor,
 180         PUSHORT NodeNumber );
 181 typedef VOID (WINAPI *func_GetCurrentProcessorNumberEx_t)(
 182         PPROCESSOR_NUMBER ProcNumber );
 183
 184 typedef HANDLE (WINAPI *func_CreateRemoteThreadEx_t)(
 185         HANDLE hProcess,
 186         LPSECURITY_ATTRIBUTES lpThreadAttributes,
 187         SIZE_T dwStackSize,
 188         LPTHREAD_START_ROUTINE lpStartAddress,
 189         LPVOID lpParameter,
 190         DWORD dwCreationFlags,
 191         LPPROC_THREAD_ATTRIBUTE_LIST lpAttributeList,
 192         LPDWORD lpThreadId);
 193
 194 typedef BOOL (WINAPI *func_InitializeProcThreadAttributeList_t)(
 195         LPPROC_THREAD_ATTRIBUTE_LIST lpAttributeList,
 196         DWORD dwAttributeCount,
 197         DWORD dwFlags,
 198         PSIZE_T lpSize);
 199 typedef BOOL (WINAPI *func_UpdateProcThreadAttribute_t)(
 200         LPPROC_THREAD_ATTRIBUTE_LIST lpAttributeList,
 201         DWORD dwFlags,
 202         DWORD_PTR Attribute,
 203         PVOID lpValue,
 204         SIZE_T cbSize,
 205         PVOID lpPreviousValue,
 206         PSIZE_T lpReturnSize);
 207 typedef VOID (WINAPI *func_DeleteProcThreadAttributeList_t)(
 208         LPPROC_THREAD_ATTRIBUTE_LIST lpAttributeList);
 209 typedef DWORD (WINAPI *func_GetActiveProcessorCount_t)(WORD GroupNumber);
 210 typedef WORD (WINAPI *func_GetActiveProcessorGroupCount_t)(void);
 211
 212
 213 /* WinXP SP2, WinXP64, WinSrv 2003 */
 214 func_GetNumaHighestNodeNumber_t             func_GetNumaHighestNodeNumber;
 215 func_SetThreadIdealProcessor_t              func_SetThreadIdealProcessor;
 216 /* Windows 7, WinSrv 2008R2 */
 217 func_SetThreadGroupAffinity_t               func_SetThreadGroupAffinity;
 218 func_SetThreadIdealProcessorEx_t            func_SetThreadIdealProcessorEx;
 219 func_GetNumaNodeProcessorMaskEx_t           func_GetNumaNodeProcessorMaskEx;
 220 func_GetNumaProcessorNodeEx_t               func_GetNumaProcessorNodeEx;
 221 func_GetCurrentProcessorNumberEx_t          func_GetCurrentProcessorNumberEx;
 222 func_GetActiveProcessorCount_t              func_GetActiveProcessorCount;
 223 func_GetActiveProcessorGroupCount_t         func_GetActiveProcessorGroupCount;
 224 func_CreateRemoteThreadEx_t                 func_CreateRemoteThreadEx;
 225 /* Windows Vista, WinSrv 2008 */
 226 func_InitializeProcThreadAttributeList_t    func_InitializeProcThreadAttributeList;
 227 func_UpdateProcThreadAttribute_t            func_UpdateProcThreadAttribute;
 228 func_DeleteProcThreadAttributeList_t        func_DeleteProcThreadAttributeList;
 229
 230
 231
 232 /*  returns 0 on success.
 233     Success is returned if the system is non-NUMA, OR the system doesn't
 234     support appropriate NUMA APIs, OR the system is NUMA and we successfully
 235     initialized support.
 236
 237     returns -1 on error.
 238     This can happen if an API returned an error, a memory allocation failed, or
 239     we failed to initialize affinity mapping information.
 240  */
 241 int tMPI_Init_NUMA(void)
 242 {
 243     /* module handle to kernel32.dll -- we already reference it, so it's already loaded */
 244     HMODULE hModKernel32 = NULL;
 245     /* 0-based NUMA node count -- does not imply all nodes have available (eg: hot-plug) processors */
 246     ULONG   ulHighestNumaNodeNumber;
 247     /* total number of processors available per affinity masks */
 248     DWORD   dwTotalProcessors = 0;
 249     ULONG   i                 = 0;
 250
 251     /* calling thread PROCESSOR_NUMBER */
 252     PROCESSOR_NUMBER CurrentProcessorNumber;
 253     /* calling thread GROUP_AFFINITY */
 254     /*GROUP_AFFINITY CurrentThreadGroupAffinity; */
 255     /* calling thread NUMA node */
 256     /*USHORT CurrentNumaNodeNumber;*/
 257
 258     WORD wActiveGroupCount;
 259     WORD GroupIndex;
 260
 261     /* array of processor information structures */
 262     MPI_NUMA_PROCESSOR_INFO *pMPI_ProcessorInfo = NULL;
 263
 264     /* assume an error condition */
 265     int iRet = -1;
 266
 267     hModKernel32 = GetModuleHandleA("kernel32.dll");
 268
 269     if (hModKernel32 == NULL)
 270     {
 271         return 0;
 272     }
 273
 274     /* obtain addresses of relevant NUMA functions, most of which are
 275        Windows 7 / Windows Server 2008R2 only functions
 276        this is done using GetProcAddress to enable the binary to run on older
 277        Windows versions.
 278      */
 279
 280     func_GetNumaHighestNodeNumber = (func_GetNumaHighestNodeNumber_t) GetProcAddress( hModKernel32, "GetNumaHighestNodeNumber" );
 281     func_SetThreadIdealProcessor  = (func_SetThreadIdealProcessor_t) GetProcAddress( hModKernel32, "SetThreadIdealProcessor" );
 282
 283     if (func_GetNumaHighestNodeNumber == NULL)
 284     {
 285         return 0;
 286     }
 287
 288     /* determine if we're on a NUMA system and if so, determine the number of
 289        (potential) nodes */
 290
 291     if (!func_GetNumaHighestNodeNumber( &ulHighestNumaNodeNumber ))
 292     {
 293         return -1;
 294     }
 295
 296
 297
 298     func_SetThreadGroupAffinity            = (func_SetThreadGroupAffinity_t)GetProcAddress( hModKernel32, "SetThreadGroupAffinity" );
 299     func_SetThreadIdealProcessorEx         = (func_SetThreadIdealProcessorEx_t)GetProcAddress( hModKernel32, "SetThreadIdealProcessorEx" );
 300     func_CreateRemoteThreadEx              = (func_CreateRemoteThreadEx_t)GetProcAddress( hModKernel32, "CreateRemoteThreadEx" );
 301     func_GetNumaNodeProcessorMaskEx        = (func_GetNumaNodeProcessorMaskEx_t)GetProcAddress( hModKernel32, "GetNumaNodeProcessorMaskEx" );
 302     func_GetNumaProcessorNodeEx            = (func_GetNumaProcessorNodeEx_t)GetProcAddress( hModKernel32, "GetNumaProcessorNodeEx" );
 303     func_GetCurrentProcessorNumberEx       = (func_GetCurrentProcessorNumberEx_t)GetProcAddress( hModKernel32, "GetCurrentProcessorNumberEx" );
 304     func_GetActiveProcessorCount           = (func_GetActiveProcessorCount_t)GetProcAddress( hModKernel32, "GetActiveProcessorCount" );
 305     func_GetActiveProcessorGroupCount      = (func_GetActiveProcessorGroupCount_t)GetProcAddress( hModKernel32, "GetActiveProcessorGroupCount" );
 306     func_InitializeProcThreadAttributeList = (func_InitializeProcThreadAttributeList_t)GetProcAddress( hModKernel32, "InitializeProcThreadAttributeList" );
 307     func_UpdateProcThreadAttribute         = (func_UpdateProcThreadAttribute_t)GetProcAddress( hModKernel32, "UpdateProcThreadAttribute" );
 308     func_DeleteProcThreadAttributeList     = (func_DeleteProcThreadAttributeList_t)GetProcAddress( hModKernel32, "DeleteProcThreadAttributeList" );
 309
 310     if ( (func_SetThreadGroupAffinity == NULL) ||
 311          (func_SetThreadIdealProcessorEx == NULL) ||
 312          (func_CreateRemoteThreadEx == NULL) ||
 313          (func_GetNumaNodeProcessorMaskEx == NULL) ||
 314          (func_GetNumaProcessorNodeEx == NULL) ||
 315          (func_GetCurrentProcessorNumberEx == NULL) ||
 316          (func_GetActiveProcessorCount == NULL) ||
 317          (func_GetActiveProcessorGroupCount == NULL) ||
 318          (func_InitializeProcThreadAttributeList == NULL) ||
 319          (func_UpdateProcThreadAttribute == NULL) ||
 320          (func_DeleteProcThreadAttributeList == NULL) )
 321     {
 322         /* if any addresses couldn't be located, assume NUMA functionality
 323            isn't supported */
 324         return 0;
 325     }
 326 #if 0
 327     if (ulHighestNumaNodeNumber == 0)
 328     {
 329         /* system is not NUMA */
 330         return 0;
 331     }
 332 #endif
 333
 334     /* count the active processors across the groups */
 335
 336     func_GetCurrentProcessorNumberEx(&CurrentProcessorNumber);
 337
 338     wActiveGroupCount = func_GetActiveProcessorGroupCount();
 339
 340     dwTotalProcessors = func_GetActiveProcessorCount( ALL_PROCESSOR_GROUPS );
 341
 342 #if !((defined WIN64 || defined _WIN64))
 343     /* WOW64 doesn't allow setting the affinity correctly beyond 32
 344        processors -- the KAFFINITY mask is only 32 bits wide
 345        This check is only here for completeness -- large systems should be
 346        running 64bit Gromacs code, where the processor quantity is not
 347        constrained.
 348        By failing here, the WOW64 32bit client will use normal CreateThread(),
 349        which can schedule up to 64 un-affinitized threads
 350      */
 351
 352     if (dwTotalProcessors > 32)
 353     {
 354         return 0;
 355     }
 356 #endif
 357
 358     /* allocate array of processor info blocks */
 359
 360     pMPI_ProcessorInfo = malloc( sizeof(MPI_NUMA_PROCESSOR_INFO) *
 361                                  dwTotalProcessors );
 362     if (pMPI_ProcessorInfo == NULL)
 363     {
 364         goto cleanup;
 365     }
 366
 367     /* zero fill to cover reserved must be-zero fields */
 368     memset(pMPI_ProcessorInfo, 0, sizeof(MPI_NUMA_PROCESSOR_INFO) * dwTotalProcessors);
 369
 370     /* loop through each processor group, and for each group, capture the
 371        processor numbers and NUMA node information. */
 372
 373     for (GroupIndex = 0; GroupIndex < wActiveGroupCount; GroupIndex++)
 374     {
 375         DWORD dwGroupProcessorCount;
 376         BYTE  ProcessorIndex;
 377
 378         dwGroupProcessorCount = func_GetActiveProcessorCount( GroupIndex );
 379
 380         for (ProcessorIndex = 0; ProcessorIndex < dwGroupProcessorCount;
 381              ProcessorIndex++)
 382         {
 383             PROCESSOR_NUMBER *pProcessorNumber = &(pMPI_ProcessorInfo[i].ProcessorNumber);
 384             GROUP_AFFINITY   *pGroupAffinity   = &(pMPI_ProcessorInfo[i].GroupAffinity);
 385             USHORT           *pNodeNumber      = &(pMPI_ProcessorInfo[i].NumaNodeNumber);
 386
 387             pProcessorNumber->Group  = GroupIndex;
 388             pProcessorNumber->Number = ProcessorIndex;
 389
 390             /* save an index to the processor array entry for the current processor
 391                this is used to enable subsequent threads to be created in a round
 392                robin fashion starting at the next array entry
 393              */
 394
 395             if ( (CurrentProcessorNumber.Group == pProcessorNumber->Group ) &&
 396                  (CurrentProcessorNumber.Number == pProcessorNumber->Number) )
 397             {
 398                 /* set global: current thread index into processor array */
 399                 g_ulThreadIndex = i;
 400             }
 401
 402             /* capture the node number and group affinity associated with processor entry
 403                any failures here are assumed to be catastrophic and disable
 404                the group & NUMA aware thread support
 405              */
 406
 407             if (!func_GetNumaProcessorNodeEx(pProcessorNumber, pNodeNumber))
 408             {
 409                 goto cleanup;
 410             }
 411
 412             if (!func_GetNumaNodeProcessorMaskEx(*pNodeNumber, pGroupAffinity))
 413             {
 414                 goto cleanup;
 415             }
 416
 417             /* future enhancement: construct GroupAffinity (single) processor
 418                mask within NUMA node for this processor entry */
 419
 420             /* increment processor array index */
 421             i++;
 422
 423             /* sanity check, should never happen */
 424
 425             if (i > dwTotalProcessors)
 426             {
 427                 goto cleanup;
 428             }
 429         }
 430     }
 431
 432
 433     /* capture number of processors, highest NUMA node number, and processor
 434        array */
 435     g_ulTotalProcessors       = dwTotalProcessors;
 436     g_ulHighestNumaNodeNumber = ulHighestNumaNodeNumber;
 437     g_MPI_ProcessorInfo       = pMPI_ProcessorInfo;
 438
 439     iRet = 0;
 440
 441 cleanup:
 442
 443     if (iRet != 0)
 444     {
 445         if (pMPI_ProcessorInfo)
 446         {
 447             tMPI_Free( pMPI_ProcessorInfo );
 448         }
 449     }
 450
 451     return 0;
 452 }
 453
 454 static int tMPI_Thread_id_list_init(void)
 455 {
 456     int ret = 0;
 457
 458     EnterCriticalSection( &thread_id_list_lock );
 459
 460     N_thread_id_list      = 0;
 461     Nalloc_thread_id_list = 4; /* number of initial allocation*/
 462     thread_id_list        = (thread_id_list_t*)malloc(sizeof(thread_id_list_t)*
 463                                                       Nalloc_thread_id_list);
 464     if (thread_id_list == NULL)
 465     {
 466         ret = ENOMEM;
 467     }
 468
 469     LeaveCriticalSection( &thread_id_list_lock );
 470     return ret;
 471 }
 472
 473
 474 /* add an entry to the thread ID list, assuming it's locked */
 475 static int tMPI_Thread_id_list_add_locked(DWORD               thread_id,
 476                                           struct tMPI_Thread *th)
 477 {
 478     if (Nalloc_thread_id_list < N_thread_id_list + 1)
 479     {
 480         thread_id_list_t* new_list;
 481         int               i;
 482
 483         /* double the size */
 484         Nalloc_thread_id_list *= 2;
 485         /* and allocate the new list */
 486         new_list = (thread_id_list_t*)malloc(sizeof(thread_id_list_t)*
 487                                              Nalloc_thread_id_list);
 488         if (new_list == NULL)
 489         {
 490             return ENOMEM;
 491         }
 492         /* and copy over all elements */
 493         for (i = 0; i < N_thread_id_list; i++)
 494         {
 495             new_list[i] = thread_id_list[i];
 496         }
 497         /* free the old list */
 498         tMPI_Free(thread_id_list);
 499         thread_id_list = new_list;
 500     }
 501     thread_id_list[ N_thread_id_list ].thread_id = thread_id;
 502     thread_id_list[ N_thread_id_list ].th        = th;
 503     N_thread_id_list++;
 504
 505     return 0;
 506 }
 507
 508
 509 /* add an entry to the thread ID list */
 510 static int tMPI_Thread_id_list_add(DWORD thread_id, struct tMPI_Thread *th)
 511 {
 512     int ret = 0;
 513     EnterCriticalSection( &thread_id_list_lock );
 514     ret = tMPI_Thread_id_list_add_locked(thread_id, th);
 515     LeaveCriticalSection( &thread_id_list_lock );
 516     return ret;
 517 }
 518
 519 /* Remove an entry from the thread_id list, assuming it's locked.
 520    Does nothing if an entry is not found.*/
 521 static void tMPI_Thread_id_list_remove_locked(DWORD thread_id)
 522 {
 523     int       i;
 524     tmpi_bool found = FALSE;
 525
 526     /* move the last thread_id_list item to the one we want to remove */
 527     for (i = 0; i < N_thread_id_list; i++)
 528     {
 529         if (thread_id_list[i].thread_id == thread_id)
 530         {
 531             thread_id_list[i] = thread_id_list[N_thread_id_list - 1];
 532             found             = TRUE;
 533             break;
 534         }
 535     }
 536
 537     if (found)
 538     {
 539         N_thread_id_list--;
 540     }
 541 }
 542
 543
 544 /* Remove an entry from the thread_id list */
 545 static void tMPI_Thread_id_list_remove(DWORD thread_id)
 546 {
 547
 548     EnterCriticalSection( &thread_id_list_lock );
 549     tMPI_Thread_id_list_remove_locked(thread_id);
 550     LeaveCriticalSection( &thread_id_list_lock );
 551 }
 552
 553
 554
 555 /* try to find a thread id in the thread id list. Return NULL when there is no
 556    such thread id in the list. Assumes the list is locked.*/
 557 static struct tMPI_Thread *tMPI_Thread_id_list_find_locked(DWORD thread_id)
 558 {
 559     int                 i;
 560     struct tMPI_Thread *ret = NULL;
 561
 562     /* this is a linear search but it's only O(Nthreads). */
 563     for (i = 0; i < N_thread_id_list; i++)
 564     {
 565         if (thread_id_list[i].thread_id == thread_id)
 566         {
 567             ret = thread_id_list[i].th;
 568             break;
 569         }
 570     }
 571
 572     return ret;
 573 }
 574
 575 /* try to find a thread id in the thread id list. Return NULL when there is no
 576    such thread id in the list.*/
 577 static struct tMPI_Thread *tMPI_Thread_id_list_find(DWORD thread_id)
 578 {
 579     struct tMPI_Thread *ret = NULL;
 580
 581     EnterCriticalSection( &thread_id_list_lock );
 582     ret = tMPI_Thread_id_list_find_locked(thread_id);
 583     LeaveCriticalSection( &thread_id_list_lock );
 584     return ret;
 585 }
 586
 587 /* try to add the running thread to the list. Returns the tMPI_Thrread struct
 588    associated with this thread, or NULL in case of an error.*/
 589 static struct tMPI_Thread *tMPI_Thread_id_list_add_self(void)
 590 {
 591     DWORD               thread_id;
 592     struct tMPI_Thread *th = NULL;
 593     int                 ret;
 594
 595     EnterCriticalSection( &thread_id_list_lock );
 596
 597     thread_id = GetCurrentThreadId();
 598     th        = tMPI_Thread_id_list_find_locked(thread_id);
 599     if (th == NULL)
 600     {
 601         /* if not, create an ID, set it and return it */
 602         th = (struct tMPI_Thread*)malloc(sizeof(struct tMPI_Thread)*1);
 603
 604         /* to create a handle that can be used outside of the current
 605            thread, the handle from GetCurrentThread() must first
 606            be duplicated.. */
 607         DuplicateHandle(GetCurrentProcess(),
 608                         GetCurrentThread(),
 609                         GetCurrentProcess(),
 610                         &th->th,
 611                         0,
 612                         FALSE,
 613                         DUPLICATE_SAME_ACCESS);
 614
 615         /* This causes a small memory leak that is hard to fix. */
 616         th->started_by_tmpi = 0;
 617         ret                 = tMPI_Thread_id_list_add_locked(thread_id, th);
 618         if (ret != 0)
 619         {
 620             free(th);
 621             th = NULL;
 622         }
 623     }
 624     LeaveCriticalSection( &thread_id_list_lock );
 625     return th;
 626 }
 627
 628
 629 static int tMPI_Init_initers(void)
 630 {
 631     int state;
 632     int ret = 0;
 633
 634     /* we can pre-check because it's atomic */
 635     if (tMPI_Atomic_get(&init_inited) == 0)
 636     {
 637         /* this can be a spinlock because the chances of collision are low. */
 638         tMPI_Spinlock_lock( &init_init );
 639
 640         state = tMPI_Atomic_get(&init_inited);
 641         tMPI_Atomic_memory_barrier_acq();
 642         if (state == 0)
 643         {
 644             InitializeCriticalSection(&mutex_init);
 645             InitializeCriticalSection(&once_init);
 646             InitializeCriticalSection(&cond_init);
 647             InitializeCriticalSection(&barrier_init);
 648             InitializeCriticalSection(&thread_id_list_lock);
 649
 650             ret = tMPI_Init_NUMA();
 651             if (ret != 0)
 652             {
 653                 goto err;
 654             }
 655
 656
 657             ret = tMPI_Thread_id_list_init();
 658             if (ret != 0)
 659             {
 660                 goto err;
 661             }
 662
 663             tMPI_Atomic_memory_barrier_rel();
 664             tMPI_Atomic_set(&init_inited, 1);
 665         }
 666
 667         tMPI_Spinlock_unlock( &init_init );
 668     }
 669     return ret;
 670 err:
 671     tMPI_Spinlock_unlock( &init_init );
 672     return ret;
 673 }
 674
 675
 676
 677 enum tMPI_Thread_support tMPI_Thread_support(void)
 678 {
 679     return TMPI_THREAD_SUPPORT_YES;
 680 }
 681
 682 struct tMPI_Thread_starter_param
 683 {
 684     void               *(*start_routine)(void*); /* the function */
 685     void               *param;                   /* its parameter */
 686     struct tMPI_Thread *thread;
 687 };
 688
 689 static DWORD WINAPI tMPI_Win32_thread_starter( LPVOID lpParam )
 690 {
 691     struct tMPI_Thread_starter_param *prm =
 692         (struct tMPI_Thread_starter_param*)lpParam;
 693
 694     (prm->start_routine)(prm->param);
 695     return 0;
 696 }
 697
 698
 699 int tMPI_Thread_get_hw_number(void)
 700 {
 701     int         ret;
 702
 703     SYSTEM_INFO sysinfo;
 704     GetSystemInfo( &sysinfo );
 705
 706     ret = sysinfo.dwNumberOfProcessors;
 707     return ret;
 708 }
 709
 710
 711
 712
 713 int tMPI_Thread_create(tMPI_Thread_t *thread,
 714                        void *(*start_routine)(void *), void *arg)
 715 {
 716     DWORD thread_id;
 717     struct tMPI_Thread_starter_param *prm;
 718     int   ret;
 719
 720     ret = tMPI_Init_initers();
 721     if (ret != 0)
 722     {
 723         return ret;
 724     }
 725
 726     if (thread == NULL)
 727     {
 728         return EINVAL;
 729     }
 730
 731     /* a small memory leak to be sure that it doesn't get deallocated
 732        once this function ends, before the newly created thread uses it. */
 733     prm = (struct tMPI_Thread_starter_param*)
 734         malloc(sizeof(struct tMPI_Thread_starter_param));
 735     if (prm == NULL)
 736     {
 737         return ENOMEM;
 738     }
 739     prm->start_routine = start_routine;
 740     prm->param         = arg;
 741
 742     *thread = (struct tMPI_Thread*)malloc(sizeof(struct tMPI_Thread)*1);
 743     if (*thread == NULL)
 744     {
 745         free(prm);
 746         return ENOMEM;
 747     }
 748
 749     /* this must be locked before the thread is created to prevent a race
 750        condition if the thread immediately wants to create its own entry */
 751     EnterCriticalSection( &thread_id_list_lock );
 752     /* just create a plain thread. */
 753     (*thread)->started_by_tmpi = 1;
 754     (*thread)->th              = CreateThread(NULL,
 755                                               0,
 756                                               tMPI_Win32_thread_starter,
 757                                               prm,
 758                                               0,
 759                                               &thread_id);
 760     if ((*thread)->th == NULL)
 761     {
 762         ret = -1;
 763         goto err;
 764     }
 765     (*thread)->id = thread_id;
 766
 767     if ((*thread)->th == NULL)
 768     {
 769         ret = -1;
 770         goto err;
 771     }
 772     ret = tMPI_Thread_id_list_add_locked(thread_id, (*thread));
 773     if (ret != 0)
 774     {
 775         goto err;
 776     }
 777     LeaveCriticalSection( &thread_id_list_lock );
 778
 779 #if 0
 780     /* inherit the thread priority from the parent thread. */
 781     /* TODO: is there value in setting this, vs. just allowing it to default
 782        from the process?  currently, this limits the effectivenes of changing
 783        the priority in eg: TaskManager. */
 784     SetThreadPriority(((*thread)->th), GetThreadPriority(GetCurrentThread()));
 785 #endif
 786
 787     return 0;
 788 err:
 789     free(prm);
 790     free(thread);
 791     LeaveCriticalSection( &thread_id_list_lock );
 792     return ret;
 793 }
 794
 795
 796
 797
 798
 799
 800
 801 int tMPI_Thread_join(tMPI_Thread_t thread, void **value_ptr)
 802 {
 803     DWORD ret, retval;
 804
 805     ret = WaitForSingleObject(thread->th, INFINITE);
 806     if (ret != 0)
 807     {
 808         return -1;
 809     }
 810
 811     if (value_ptr)
 812     {
 813         if (!GetExitCodeThread(thread, &retval))
 814         {
 815             return -1;
 816         }
 817     }
 818     CloseHandle(thread->th);
 819     tMPI_Thread_id_list_remove(thread->id);
 820     free(thread);
 821
 822     return 0;
 823 }
 824
 825
 826 void tMPI_Thread_exit(void *value_ptr)
 827 {
 828     /* TODO: call destructors for thread-local storage */
 829     ExitThread( 0 );
 830 }
 831
 832
 833
 834
 835 int tMPI_Thread_cancel(tMPI_Thread_t thread)
 836 {
 837     if (!TerminateThread( thread, -1) )
 838     {
 839         return -1;
 840     }
 841     tMPI_Thread_id_list_remove(thread->id);
 842     return 0;
 843 }
 844
 845
 846 tMPI_Thread_t tMPI_Thread_self(void)
 847 {
 848     tMPI_Thread_t th;
 849     int           ret;
 850
 851     ret = tMPI_Init_initers();
 852     if (ret != 0)
 853     {
 854         return NULL;
 855     }
 856
 857     th = tMPI_Thread_id_list_add_self();
 858     return th;
 859 }
 860
 861
 862 int tMPI_Thread_equal(tMPI_Thread_t t1, tMPI_Thread_t t2)
 863 {
 864     /* because the tMPI thread IDs are unique, we can compare them directly */
 865     return (t1 == t2);
 866 }
 867
 868 enum tMPI_Thread_setaffinity_support tMPI_Thread_setaffinity_support(void)
 869 {
 870     /* Windows supports seting of thread affinities */
 871     return TMPI_SETAFFINITY_SUPPORT_YES;
 872 }
 873
 874 int tMPI_Thread_setaffinity_single(tMPI_Thread_t thread, unsigned int nr)
 875 {
 876     GROUP_AFFINITY   GroupAffinity;
 877     PROCESSOR_NUMBER IdealProcessorNumber;
 878     /* thread NUMA node */
 879     USHORT           NumaNodeNumber;
 880
 881     /* check for a processor info array. This exists if NUMA
 882        style calls have been succesfully initialized. */
 883     if (g_MPI_ProcessorInfo != NULL)
 884     {
 885
 886         /*func_GetCurrentProcessorNumberEx(&CurrentProcessorNumber);*/
 887         /* group, mask. */
 888         memcpy(&GroupAffinity,
 889                &(g_MPI_ProcessorInfo[nr].GroupAffinity),
 890                sizeof(GROUP_AFFINITY));
 891
 892         /* group, processor number */
 893
 894         memcpy(&IdealProcessorNumber,
 895                &(g_MPI_ProcessorInfo[nr].ProcessorNumber),
 896                sizeof(PROCESSOR_NUMBER));
 897
 898
 899         /* set the NUMA node affinity for the current thread
 900            failures to set the current thread affinity are ignored,
 901            as a fringe case can arise on >32 processor systems with a 32bit
 902            build/code.
 903          */
 904         func_SetThreadIdealProcessorEx(thread->th,
 905                                        &IdealProcessorNumber,
 906                                        NULL);
 907
 908         if (func_GetNumaProcessorNodeEx(&IdealProcessorNumber,
 909                                         &NumaNodeNumber))
 910         {
 911             /* for the NUMA node number associated with the current processor
 912                number, get the group affinity mask */
 913             if (func_GetNumaNodeProcessorMaskEx(NumaNodeNumber,
 914                                                 &GroupAffinity))
 915             {
 916                 /* set the current thread affinity to prevent it from running
 917                    on other NUMA nodes */
 918                 func_SetThreadGroupAffinity(thread->th,
 919                                             &GroupAffinity,
 920                                             NULL);
 921                 return 0;
 922             }
 923         }
 924         return 1;
 925     }
 926     else
 927     {
 928         /* No NUMA-style calls. We just do a simpler thing. */
 929         if ( (func_SetThreadIdealProcessor != NULL) )
 930         {
 931             return (func_SetThreadIdealProcessor(thread->th, nr) == -1);
 932         }
 933     }
 934     return 0;
 935 }
 936
 937
 938
 939 int tMPI_Thread_mutex_init(tMPI_Thread_mutex_t *mtx)
 940 {
 941     if (mtx == NULL)
 942     {
 943         return EINVAL;
 944     }
 945
 946     mtx->mutex = (struct tMPI_Mutex*)malloc(sizeof(struct tMPI_Mutex)*1);
 947     if (mtx->mutex == NULL)
 948     {
 949         return ENOMEM;
 950     }
 951     InitializeCriticalSection(&(mtx->mutex->cs));
 952
 953     return 0;
 954 }
 955
 956
 957 int tMPI_Thread_mutex_destroy(tMPI_Thread_mutex_t *mtx)
 958 {
 959     if (mtx == NULL)
 960     {
 961         return EINVAL;
 962     }
 963
 964     DeleteCriticalSection(&(mtx->mutex->cs));
 965     free(mtx->mutex);
 966
 967     return 0;
 968 }
 969
 970
 971
 972
 973 static int tMPI_Thread_mutex_init_once(tMPI_Thread_mutex_t *mtx)
 974 {
 975     int ret = 0;
 976
 977     /* This is essentially a copy of the code from the one-time
 978      * initialization, but with a call to the mutex init routine instead.
 979      * It might seem like overkill, but it will only be executed the first
 980      * time you call a static mutex, and it is important to get all the
 981      * memory barriers right. Trust me, you don't want a deadlock here...
 982      */
 983
 984     /* initialize the initializers */
 985     ret = tMPI_Init_initers();
 986     if (ret != 0)
 987     {
 988         return ret;
 989     }
 990     /* Lock the common one-time init mutex so we can check carefully */
 991     EnterCriticalSection( &mutex_init );
 992
 993     /* Do the actual (locked) check - system mutex is locked if we get here */
 994     if (mtx->mutex == NULL)
 995     {
 996         /* No need to keep the lock during execution -
 997          * Only one thread can do it anyway.
 998          */
 999         ret = tMPI_Thread_mutex_init(mtx);
1000     }
1001     LeaveCriticalSection( &mutex_init );
1002
1003     return ret;
1004 }
1005
1006
1007
1008 int tMPI_Thread_mutex_lock(tMPI_Thread_mutex_t *mtx)
1009 {
1010     /* check whether the mutex is initialized */
1011     if (tMPI_Atomic_get( &(mtx->initialized)  ) == 0)
1012     {
1013         tMPI_Thread_mutex_init_once(mtx);
1014     }
1015
1016     /* The mutex is now guaranteed to be valid. */
1017     EnterCriticalSection( &(mtx->mutex->cs) );
1018
1019     return 0;
1020 }
1021
1022
1023
1024
1025 int tMPI_Thread_mutex_trylock(tMPI_Thread_mutex_t *mtx)
1026 {
1027     BOOL ret;
1028
1029     /* check whether the mutex is initialized */
1030     if (tMPI_Atomic_get( &(mtx->initialized)  ) == 0)
1031     {
1032         tMPI_Thread_mutex_init_once(mtx);
1033     }
1034
1035     /* The mutex is now guaranteed to be valid. */
1036     ret = TryEnterCriticalSection( &(mtx->mutex->cs) );
1037
1038     return (ret != 0);
1039 }
1040
1041
1042
1043 int tMPI_Thread_mutex_unlock(tMPI_Thread_mutex_t *mtx)
1044 {
1045     /* we should have initialized our critical section anyway */
1046     LeaveCriticalSection( &(mtx->mutex->cs) );
1047
1048     return 0;
1049 }
1050
1051
1052
1053 int tMPI_Thread_key_create(tMPI_Thread_key_t *key, void (*destructor)(void *))
1054 {
1055     if (key == NULL)
1056     {
1057         return EINVAL;
1058     }
1059
1060
1061     /* TODO: make list of destructors for thread-local storage */
1062     key->key = (struct tMPI_Thread_key*)malloc(sizeof(struct tMPI_Thread_key));
1063     if (key->key == NULL)
1064     {
1065         return ENOMEM;
1066     }
1067
1068     (key)->key->wkey = TlsAlloc();
1069
1070     if ( (key)->key->wkey == TLS_OUT_OF_INDEXES)
1071     {
1072         return -1;
1073     }
1074
1075     return 0;
1076 }
1077
1078
1079 int tMPI_Thread_key_delete(tMPI_Thread_key_t key)
1080 {
1081     TlsFree(key.key->wkey);
1082     free(key.key);
1083
1084     return 0;
1085 }
1086
1087
1088
1089 void * tMPI_Thread_getspecific(tMPI_Thread_key_t key)
1090 {
1091     void *p = NULL;
1092
1093     p = TlsGetValue(key.key->wkey);
1094
1095     return p;
1096 }
1097
1098
1099 int tMPI_Thread_setspecific(tMPI_Thread_key_t key, void *value)
1100 {
1101     BOOL ret;
1102
1103     ret = TlsSetValue(key.key->wkey, value);
1104
1105     return ret == 0;
1106 }
1107
1108 #if 0
1109 /* use once Vista is minimum required version */
1110 static BOOL CALLBACK InitHandleWrapperFunction(PINIT_ONCE InitOnce,
1111                                                PVOID      Parameter,
1112                                                PVOID     *lpContext)
1113 {
1114     void (*fn)(void) = (void (*)(void))Parameter;
1115
1116     fn();
1117
1118     return TRUE;
1119 }
1120
1121 CRITICAL_SECTION tMPI_Once_cs;
1122 tMPI_Spinlock_t  tMPI_Once_cs_lock = TMPI_SPINLOCK_INITIALIZER;
1123 volatile int     tMPI_Once_init    = 0;
1124 #endif
1125
1126 int tMPI_Thread_once(tMPI_Thread_once_t *once_control,
1127                      void                (*init_routine)(void))
1128 {
1129 #if 0
1130     /* use once Vista is minimum required version */
1131     BOOL bStatus;
1132     bStatus = InitOnceExecuteOnce(once_control, InitHandleWrapperFunction,
1133                                   init_routine, NULL);
1134
1135     if (!bStatus)
1136     {
1137         return -1;
1138     }
1139 #else
1140     int ret;
1141
1142     /* really ugly hack - and it's slow... */
1143     ret = tMPI_Init_initers();
1144     if (ret != 0)
1145     {
1146         return ret;
1147     }
1148
1149     EnterCriticalSection(&once_init);
1150     if (tMPI_Atomic_get(&(once_control->once)) == 0)
1151     {
1152         (*init_routine)();
1153         tMPI_Atomic_set(&(once_control->once), 1);
1154     }
1155     LeaveCriticalSection(&once_init);
1156 #endif
1157     return 0;
1158 }
1159
1160
1161
1162
1163
1164 int tMPI_Thread_cond_init(tMPI_Thread_cond_t *cond)
1165 {
1166     if (cond == NULL)
1167     {
1168         return EINVAL;
1169     }
1170
1171     cond->condp = (struct tMPI_Thread_cond*)
1172         malloc(sizeof(struct tMPI_Thread_cond));
1173     if (cond->condp == NULL)
1174     {
1175         return ENOMEM;
1176     }
1177 #if 0
1178     /* use this code once Vista is the minimum version required */
1179     InitializeConditionVariable( &(cond->cv) );
1180 #else
1181     cond->condp->Nwaiters = 0;
1182     InitializeCriticalSection(&(cond->condp->wtr_lock));
1183     cond->condp->Nrelease = 0;
1184     cond->condp->cycle    = 0;
1185     /* a manual reset, unsignalled event */
1186     cond->condp->ev = CreateEvent(NULL, TRUE, FALSE, NULL);
1187 #endif
1188     return 0;
1189 }
1190
1191
1192 int tMPI_Thread_cond_destroy(tMPI_Thread_cond_t *cond)
1193 {
1194 #if 0
1195     /* use this code once Vista is the minimum version required */
1196     /* windows doesnt have this function */
1197 #else
1198     DeleteCriticalSection(&(cond->condp->wtr_lock));
1199     free(cond->condp);
1200 #endif
1201     return 0;
1202 }
1203
1204
1205
1206 /*! \brief Static init routine for pthread barrier
1207  *
1208  * \internal
1209  *
1210  * This is only used as a wrapper to enable static initialization
1211  * of posix thread types together with out abstraction layer for tMPI_Thread.h
1212  *
1213  * \param cond  Condition variable, must be statically initialized
1214  *
1215  * \return status - 0 on success, or a standard error code.
1216  */
1217 static int tMPI_Thread_cond_init_once(tMPI_Thread_cond_t *cond)
1218 {
1219     int ret = 0;
1220
1221     /* This is essentially a copy of the code from the one-time
1222      * initialization, but with a call to the cond init routine instead.
1223      * It might seem like overkill, but it will only be executed the first
1224      * time you call a static condition variable, and it is important to get
1225      * the memory barriers right. Trust me, you don't want a deadlock here...
1226      */
1227
1228     /* initialize the initializers */
1229     ret = tMPI_Init_initers();
1230     if (ret != 0)
1231     {
1232         return ret;
1233     }
1234     /* Lock the common one-time init mutex so we can check carefully */
1235     EnterCriticalSection( &cond_init );
1236
1237     /* Do the actual (locked) check - system mutex is locked if we get here */
1238     if (cond->condp == NULL)
1239     {
1240         /* No need to keep the lock during execution -
1241          * Only one thread can do it anyway.  */
1242         ret = tMPI_Thread_cond_init(cond);
1243     }
1244     LeaveCriticalSection( &cond_init );
1245
1246     return ret;
1247 }
1248
1249
1250
1251
1252 int tMPI_Thread_cond_wait(tMPI_Thread_cond_t *cond, tMPI_Thread_mutex_t *mtx)
1253 {
1254     BOOL wait_done   = FALSE;
1255     BOOL last_waiter = FALSE;
1256     int  my_cycle;
1257     int  ret;
1258
1259     /* check whether the condition is initialized */
1260     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
1261     {
1262         ret = tMPI_Thread_cond_init_once(cond);
1263         if (ret != 0)
1264         {
1265             return ret;
1266         }
1267     }
1268     /* the mutex must have been initialized because it should be locked here */
1269
1270 #if 0
1271     /* use this code once Vista is the minimum version required */
1272     ret = SleepConditionVariableCS (&(cond->cv), &(mtx->cs), INFINITE);
1273
1274     if (!ret)
1275     {
1276         return -1;
1277     }
1278 #else
1279     /* serially increase waiter count */
1280     EnterCriticalSection(&(cond->condp->wtr_lock));
1281     cond->condp->Nwaiters++;
1282     my_cycle = cond->condp->cycle;
1283     LeaveCriticalSection(&(cond->condp->wtr_lock));
1284
1285     /* now it's safe to release the mutex from the fn call */
1286     LeaveCriticalSection(&(mtx->mutex->cs));
1287
1288     /* Loop a wait until we found out we've waited for the right event.
1289        Note that this loop is potentially a busy-wait loop in bad
1290        circumstances (higher priority threads, for example). */
1291     do
1292     {
1293         /* do the actual waiting */
1294         if (WaitForSingleObject( cond->condp->ev, INFINITE ) == WAIT_FAILED)
1295         {
1296             return -1;
1297         }
1298
1299         /* serially check whether we got the right event.  */
1300         EnterCriticalSection(&(cond->condp->wtr_lock));
1301         wait_done = (cond->condp->Nrelease > 0) &&
1302             (cond->condp->cycle != my_cycle);
1303         LeaveCriticalSection(&(cond->condp->wtr_lock));
1304     }
1305     while (!wait_done);
1306
1307     /* We obtain the mutex from the function call */
1308     EnterCriticalSection(&(mtx->mutex->cs));
1309
1310     /* we serially decrease the waiter count and release count */
1311     EnterCriticalSection(&(cond->condp->wtr_lock));
1312     cond->condp->Nwaiters--;
1313     cond->condp->Nrelease--;
1314     last_waiter = (cond->condp->Nrelease == 0);
1315     LeaveCriticalSection(&(cond->condp->wtr_lock));
1316
1317     /* manually release the event if everybody's done with it */
1318     if (last_waiter)
1319     {
1320         if (!ResetEvent( cond->condp->ev ))
1321         {
1322             return -1;
1323         }
1324     }
1325 #endif
1326
1327     return 0;
1328 }
1329
1330
1331
1332
1333 int tMPI_Thread_cond_signal(tMPI_Thread_cond_t *cond)
1334 {
1335     int ret;
1336     /* check whether the condition is initialized */
1337     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
1338     {
1339         ret = tMPI_Thread_cond_init_once(cond);
1340         if (ret != 0)
1341         {
1342             return ret;
1343         }
1344     }
1345     /* The condition variable is now guaranteed to be valid. */
1346 #if 0
1347     /* use this code once Vista is the minimum version required */
1348     WakeConditionVariable( &(cond->cv) );
1349 #else
1350     EnterCriticalSection(&(cond->condp->wtr_lock));
1351     /* check if we're not still busy with a release. If we are, do nothing. */
1352     if (cond->condp->Nwaiters > cond->condp->Nrelease)
1353     {
1354         cond->condp->Nrelease++;
1355         cond->condp->cycle++;
1356         if (!SetEvent(cond->condp->ev)) /* actually release the
1357                                            waiting threads */
1358         {
1359             return -1;
1360         }
1361     }
1362     LeaveCriticalSection(&(cond->condp->wtr_lock));
1363 #endif
1364
1365     return 0;
1366 }
1367
1368
1369
1370 int tMPI_Thread_cond_broadcast(tMPI_Thread_cond_t *cond)
1371 {
1372     int ret;
1373     /* check whether the condition is initialized */
1374     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
1375     {
1376         ret = tMPI_Thread_cond_init_once(cond);
1377         if (ret != 0)
1378         {
1379             return ret;
1380         }
1381
1382     }
1383     /* The condition variable is now guaranteed to be valid. */
1384 #if 0
1385     /* use this code once Vista is the minimum version required */
1386     WakeAllConditionVariable( &(cond->cv) );
1387 #else
1388     EnterCriticalSection(&(cond->condp->wtr_lock));
1389     /* check whether there are any waiters */
1390     if (cond->condp->Nwaiters > 0)
1391     {
1392         cond->condp->Nrelease = cond->condp->Nwaiters;
1393         cond->condp->cycle++;
1394         if (!SetEvent(cond->condp->ev)) /* actually release the
1395                                            waiting threads */
1396         {
1397             return -1;
1398         }
1399     }
1400     LeaveCriticalSection(&(cond->condp->wtr_lock));
1401 #endif
1402     return 0;
1403 }
1404
1405
1406
1407
1408 int tMPI_Thread_barrier_init(tMPI_Thread_barrier_t *barrier, int n)
1409 {
1410     int ret;
1411
1412     if (barrier == NULL)
1413     {
1414         return EINVAL;
1415     }
1416
1417     barrier->barrierp = (struct tMPI_Thread_barrier*)
1418         malloc(sizeof(struct tMPI_Thread_barrier)*1);
1419     if (barrier->barrierp == NULL)
1420     {
1421         return ENOMEM;
1422     }
1423
1424 #if 0
1425     /* use this once Vista is the oldest supported windows version: */
1426     InitializeCriticalSection(&(barrier->barrierp->cs));
1427     InitializeConditionVariable(&(barrier->barrierp->cv));
1428 #else
1429     ret = tMPI_Thread_mutex_init(&(barrier->barrierp->cs));
1430     if (ret != 0)
1431     {
1432         return ret;
1433     }
1434     ret = tMPI_Thread_cond_init(&(barrier->barrierp->cv));
1435     if (ret != 0)
1436     {
1437         return ret;
1438     }
1439 #endif
1440
1441     barrier->threshold = n;
1442     barrier->count     = n;
1443     barrier->cycle     = 0;
1444
1445     return 0;
1446 }
1447
1448
1449
1450 int tMPI_Thread_barrier_destroy(tMPI_Thread_barrier_t *barrier)
1451 {
1452     int ret;
1453
1454     if (barrier == NULL)
1455     {
1456         return EINVAL;
1457     }
1458
1459 #if 0
1460     DeleteCriticalSection(&(barrier->barrierp->cs));
1461 #else
1462     ret = tMPI_Thread_mutex_destroy(&(barrier->barrierp->cs));
1463     if (ret != 0)
1464     {
1465         return ret;
1466     }
1467 #endif
1468
1469     ret = tMPI_Thread_cond_destroy(&(barrier->barrierp->cv));
1470     if (ret != 0)
1471     {
1472         return ret;
1473     }
1474
1475     free(barrier->barrierp);
1476
1477     return 0;
1478 }
1479
1480
1481
1482 /*! \brief Static init routine for pthread barrier
1483  *
1484  * \internal
1485  *
1486  * This is only used as a wrapper to enable static initialization
1487  * of posix thread types together with out abstraction layer for tMPI_Thread.h
1488  *
1489  * \param barrier Statically initialized barrier type
1490  * \param n       Number of members in barrier
1491  *
1492  * \return status - 0 on success, or a standard error code.
1493  */
1494 static int tMPI_Thread_barrier_init_once(tMPI_Thread_barrier_t *barrier, int n)
1495 {
1496     int ret;
1497
1498     /* This is essentially a copy of the code from the one-time
1499      * initialization, but with a call to the cond init routine instead.
1500      * It might seem like overkill, but it will only be executed the first
1501      * time you call a static condition variable, and it is important to get
1502      * the memory barriers right. Trust me, you don't want a deadlock here...
1503      */
1504
1505
1506     /* initialize the initializers */
1507     ret = tMPI_Init_initers();
1508     if (ret != 0)
1509     {
1510         return ret;
1511     }
1512
1513     /* Lock the common one-time init mutex so we can check carefully */
1514     EnterCriticalSection( &barrier_init );
1515
1516     /* Do the actual (locked) check - system mutex is locked if we get here */
1517     if (barrier->barrierp == NULL)
1518     {
1519         /* No need to keep the lock during execution -
1520          * Only one thread can do it anyway.  */
1521         ret = tMPI_Thread_barrier_init(barrier, n);
1522     }
1523     LeaveCriticalSection( &barrier_init );
1524
1525     return ret;
1526 }
1527
1528
1529
1530 int tMPI_Thread_barrier_wait(tMPI_Thread_barrier_t *barrier)
1531 {
1532     int  cycle;
1533     BOOL rc  = FALSE;
1534     int  ret = 0;
1535     /*tMPI_Thread_pthread_barrier_t *p;*/
1536
1537     /* check whether the barrier is initialized */
1538     if (tMPI_Atomic_get( &(barrier->initialized)  ) == 0)
1539     {
1540         ret = tMPI_Thread_barrier_init_once(barrier, barrier->threshold);
1541         if (ret != 0)
1542         {
1543             return ret;
1544         }
1545     }
1546 #if 0
1547     EnterCriticalSection( &(barrier->barrierp->cs)  );
1548 #else
1549     ret = tMPI_Thread_mutex_lock( &(barrier->barrierp->cs) );
1550     if (ret != 0)
1551     {
1552         return ret;
1553     }
1554 #endif
1555
1556
1557
1558     cycle = barrier->cycle;
1559
1560     /* Decrement the count atomically and check if it is zero.
1561      * This will only be true for the last thread calling us.
1562      */
1563     if (--(barrier->count) <= 0)
1564     {
1565         barrier->cycle = !barrier->cycle;
1566         barrier->count = barrier->threshold;
1567 #if 0
1568         WakeAllConditionVariable( &(barrier->barrierp->cv) );
1569 #else
1570         ret = tMPI_Thread_cond_broadcast( &(barrier->barrierp->cv) );
1571         if (ret != 0)
1572         {
1573             return ret;
1574         }
1575 #endif
1576     }
1577     else
1578     {
1579         while (cycle == barrier->cycle)
1580         {
1581 #if 0
1582             rc = SleepConditionVariableCS (&(barrier->barrierp->cv),
1583                                            &(barrier->barrierp->cs),
1584                                            INFINITE);
1585             if (!rc)
1586             {
1587                 ret = -1;
1588                 break;
1589             }
1590 #else
1591             rc = tMPI_Thread_cond_wait(&barrier->barrierp->cv,
1592                                        &barrier->barrierp->cs);
1593             if (rc != 0)
1594             {
1595                 break;
1596             }
1597 #endif
1598         }
1599     }
1600 #if 0
1601     LeaveCriticalSection( &(barrier->barrierp->cs)  );
1602 #else
1603     tMPI_Thread_mutex_unlock( &(barrier->barrierp->cs) );
1604 #endif
1605     return ret;
1606 }
1607
1608 #else
1609
1610 /* just to have some symbols */
1611 int tMPI_Thread_winthreads = 0;
1612
1613 #endif /* THREAD_WINDOWS  */