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 = tMPI_Malloc( sizeof(MPI_NUMA_PROCESSOR_INFO) *
 361                                       dwTotalProcessors );
 362     if (pMPI_ProcessorInfo == NULL)
 363     {
 364         tMPI_Fatal_error(TMPI_FARGS, "tMPI_Malloc failed for processor information");
 365         goto cleanup;
 366     }
 367
 368     /* zero fill to cover reserved must be-zero fields */
 369     memset(pMPI_ProcessorInfo, 0, sizeof(MPI_NUMA_PROCESSOR_INFO) * dwTotalProcessors);
 370
 371     /* loop through each processor group, and for each group, capture the
 372        processor numbers and NUMA node information. */
 373
 374     for (GroupIndex = 0; GroupIndex < wActiveGroupCount; GroupIndex++)
 375     {
 376         DWORD dwGroupProcessorCount;
 377         BYTE  ProcessorIndex;
 378
 379         dwGroupProcessorCount = func_GetActiveProcessorCount( GroupIndex );
 380
 381         for (ProcessorIndex = 0; ProcessorIndex < dwGroupProcessorCount;
 382              ProcessorIndex++)
 383         {
 384             PROCESSOR_NUMBER *pProcessorNumber = &(pMPI_ProcessorInfo[i].ProcessorNumber);
 385             GROUP_AFFINITY   *pGroupAffinity   = &(pMPI_ProcessorInfo[i].GroupAffinity);
 386             USHORT           *pNodeNumber      = &(pMPI_ProcessorInfo[i].NumaNodeNumber);
 387
 388             pProcessorNumber->Group  = GroupIndex;
 389             pProcessorNumber->Number = ProcessorIndex;
 390
 391             /* save an index to the processor array entry for the current processor
 392                this is used to enable subsequent threads to be created in a round
 393                robin fashion starting at the next array entry
 394              */
 395
 396             if ( (CurrentProcessorNumber.Group == pProcessorNumber->Group ) &&
 397                  (CurrentProcessorNumber.Number == pProcessorNumber->Number) )
 398             {
 399                 /* set global: current thread index into processor array */
 400                 g_ulThreadIndex = i;
 401             }
 402
 403             /* capture the node number and group affinity associated with processor entry
 404                any failures here are assumed to be catastrophic and disable
 405                the group & NUMA aware thread support
 406              */
 407
 408             if (!func_GetNumaProcessorNodeEx(pProcessorNumber, pNodeNumber))
 409             {
 410                 tMPI_Fatal_error(TMPI_FARGS,
 411                                  "Processor enumeration, GetNumaProcessorNodeEx failed, error code=%d",
 412                                  GetLastError());
 413                 goto cleanup;
 414             }
 415
 416             if (!func_GetNumaNodeProcessorMaskEx(*pNodeNumber, pGroupAffinity))
 417             {
 418                 tMPI_Fatal_error(TMPI_FARGS,
 419                                  "Processor enumeration, GetNumaNodeProcessorMaskEx failed, error code=%d",
 420                                  GetLastError());
 421                 goto cleanup;
 422             }
 423
 424             /* future enhancement: construct GroupAffinity (single) processor
 425                mask within NUMA node for this processor entry */
 426
 427             /* increment processor array index */
 428             i++;
 429
 430             /* sanity check, should never happen */
 431
 432             if (i > dwTotalProcessors)
 433             {
 434                 tMPI_Fatal_error(TMPI_FARGS, "Processor enumeration exceeds allocated memory!");
 435                 goto cleanup;
 436             }
 437         }
 438     }
 439
 440
 441     /* capture number of processors, highest NUMA node number, and processor
 442        array */
 443     g_ulTotalProcessors       = dwTotalProcessors;
 444     g_ulHighestNumaNodeNumber = ulHighestNumaNodeNumber;
 445     g_MPI_ProcessorInfo       = pMPI_ProcessorInfo;
 446
 447     iRet = 0;
 448
 449 cleanup:
 450
 451     if (iRet != 0)
 452     {
 453         if (pMPI_ProcessorInfo)
 454         {
 455             tMPI_Free( pMPI_ProcessorInfo );
 456         }
 457     }
 458
 459     return 0;
 460 }
 461
 462 static void tMPI_Thread_id_list_init(void)
 463 {
 464     EnterCriticalSection( &thread_id_list_lock );
 465
 466     N_thread_id_list      = 0;
 467     Nalloc_thread_id_list = 4; /* number of initial allocation*/
 468     thread_id_list        = (thread_id_list_t*)tMPI_Malloc(
 469                 sizeof(thread_id_list_t)*
 470                 Nalloc_thread_id_list);
 471
 472     LeaveCriticalSection( &thread_id_list_lock );
 473 }
 474
 475
 476 /* add an entry to the thread ID list, assuming it's locked */
 477 static void tMPI_Thread_id_list_add_locked(DWORD               thread_id,
 478                                            struct tMPI_Thread *th)
 479 {
 480     if (Nalloc_thread_id_list < N_thread_id_list + 1)
 481     {
 482         thread_id_list_t* new_list;
 483         int               i;
 484
 485         /* double the size */
 486         Nalloc_thread_id_list *= 2;
 487         new_list               = (thread_id_list_t*)tMPI_Malloc(
 488                     sizeof(thread_id_list_t)*
 489                     Nalloc_thread_id_list);
 490         /* and copy over all elements */
 491         for (i = 0; i < N_thread_id_list; i++)
 492         {
 493             new_list[i] = thread_id_list[i];
 494         }
 495         /* free the old list */
 496         tMPI_Free(thread_id_list);
 497         thread_id_list = new_list;
 498     }
 499     thread_id_list[ N_thread_id_list ].thread_id = thread_id;
 500     thread_id_list[ N_thread_id_list ].th        = th;
 501     N_thread_id_list++;
 502
 503
 504 }
 505
 506
 507 /* add an entry to the thread ID list */
 508 static void tMPI_Thread_id_list_add(DWORD thread_id, struct tMPI_Thread *th)
 509 {
 510     EnterCriticalSection( &thread_id_list_lock );
 511     tMPI_Thread_id_list_add_locked(thread_id, th);
 512     LeaveCriticalSection( &thread_id_list_lock );
 513 }
 514
 515 /* Remove an entry from the thread_id list, assuming it's locked */
 516 static void tMPI_Thread_id_list_remove_locked(DWORD thread_id)
 517 {
 518     int       i;
 519     tmpi_bool found = FALSE;
 520
 521     /* move the last thread_id_list item to the one we want to remove */
 522     for (i = 0; i < N_thread_id_list; i++)
 523     {
 524         if (thread_id_list[i].thread_id == thread_id)
 525         {
 526             thread_id_list[i] = thread_id_list[N_thread_id_list - 1];
 527             found             = TRUE;
 528             break;
 529         }
 530     }
 531
 532     if (found)
 533     {
 534         N_thread_id_list--;
 535     }
 536 }
 537
 538
 539 /* Remove an entry from the thread_id list */
 540 static void tMPI_Thread_id_list_remove(DWORD thread_id)
 541 {
 542
 543     EnterCriticalSection( &thread_id_list_lock );
 544     tMPI_Thread_id_list_remove_locked(thread_id);
 545     LeaveCriticalSection( &thread_id_list_lock );
 546 }
 547
 548
 549
 550 /* try to find a thread id in the thread id list. Return NULL when there is no
 551    such thread id in the list. Assumes the list is locked.*/
 552 static struct tMPI_Thread *tMPI_Thread_id_list_find_locked(DWORD thread_id)
 553 {
 554     int                 i;
 555     struct tMPI_Thread *ret = NULL;
 556
 557     /* this is a linear search but it's only O(Nthreads). */
 558     for (i = 0; i < N_thread_id_list; i++)
 559     {
 560         if (thread_id_list[i].thread_id == thread_id)
 561         {
 562             ret = thread_id_list[i].th;
 563             break;
 564         }
 565     }
 566
 567     return ret;
 568 }
 569
 570 /* try to find a thread id in the thread id list. Return NULL when there is no
 571    such thread id in the list.*/
 572 static struct tMPI_Thread *tMPI_Thread_id_list_find(DWORD thread_id)
 573 {
 574     struct tMPI_Thread *ret = NULL;
 575
 576     EnterCriticalSection( &thread_id_list_lock );
 577     ret = tMPI_Thread_id_list_find_locked(thread_id);
 578
 579     LeaveCriticalSection( &thread_id_list_lock );
 580     return ret;
 581 }
 582
 583 /* try to add the running thread to the list. Returns the tMPI_Thrread struct
 584    associated with this thread.*/
 585 static struct tMPI_Thread *tMPI_Thread_id_list_add_self(void)
 586 {
 587     DWORD               thread_id;
 588     struct tMPI_Thread *th = NULL;
 589
 590     EnterCriticalSection( &thread_id_list_lock );
 591
 592     thread_id = GetCurrentThreadId();
 593     th        = tMPI_Thread_id_list_find_locked(thread_id);
 594     if (th == NULL)
 595     {
 596         /* if not, create an ID, set it and return it */
 597         th = (struct tMPI_Thread*)tMPI_Malloc(sizeof(struct tMPI_Thread)*1);
 598
 599         /* to create a handle that can be used outside of the current
 600            thread, the handle from GetCurrentThread() must first
 601            be duplicated.. */
 602         DuplicateHandle(GetCurrentProcess(),
 603                         GetCurrentThread(),
 604                         GetCurrentProcess(),
 605                         &th->th,
 606                         0,
 607                         FALSE,
 608                         DUPLICATE_SAME_ACCESS);
 609
 610         /* This causes a small memory leak that is hard to fix. */
 611         th->started_by_tmpi = 0;
 612         tMPI_Thread_id_list_add_locked(thread_id, th);
 613     }
 614     LeaveCriticalSection( &thread_id_list_lock );
 615
 616     return th;
 617 }
 618
 619
 620 static void tMPI_Init_initers(void)
 621 {
 622     int state;
 623     /* we can pre-check because it's atomic */
 624     if (tMPI_Atomic_get(&init_inited) == 0)
 625     {
 626         /* this can be a spinlock because the chances of collision are low. */
 627         tMPI_Spinlock_lock( &init_init );
 628
 629         state = tMPI_Atomic_get(&init_inited);
 630         tMPI_Atomic_memory_barrier_acq();
 631         if (state == 0)
 632         {
 633             InitializeCriticalSection(&mutex_init);
 634             InitializeCriticalSection(&once_init);
 635             InitializeCriticalSection(&cond_init);
 636             InitializeCriticalSection(&barrier_init);
 637             InitializeCriticalSection(&thread_id_list_lock);
 638
 639             /* fatal errors are handled by the routine by calling
 640                tMPI_Fatal_error() */
 641             tMPI_Init_NUMA();
 642
 643             tMPI_Thread_id_list_init();
 644
 645             tMPI_Atomic_memory_barrier_rel();
 646             tMPI_Atomic_set(&init_inited, 1);
 647         }
 648
 649         tMPI_Spinlock_unlock( &init_init );
 650     }
 651 }
 652
 653
 654
 655 /* TODO: this needs to go away!  (there's another one in pthreads.c)
 656    fatal errors are thankfully really rare*/
 657 void tMPI_Fatal_error(const char *file, int line, const char *message, ...)
 658 {
 659     va_list ap;
 660
 661     fprintf(stderr, "tMPI Fatal error in %s, line %d: ", file, line);
 662     va_start(ap, message);
 663     vfprintf(stderr, message, ap);
 664     va_end(ap);
 665     fprintf(stderr, "\n");
 666     abort();
 667 }
 668
 669
 670
 671 enum tMPI_Thread_support tMPI_Thread_support(void)
 672 {
 673     return TMPI_THREAD_SUPPORT_YES;
 674 }
 675
 676 struct tMPI_Thread_starter_param
 677 {
 678     void               *(*start_routine)(void*); /* the function */
 679     void               *param;                   /* its parameter */
 680     struct tMPI_Thread *thread;
 681 };
 682
 683 static DWORD WINAPI tMPI_Win32_thread_starter( LPVOID lpParam )
 684 {
 685     struct tMPI_Thread_starter_param *prm =
 686         (struct tMPI_Thread_starter_param*)lpParam;
 687
 688     (prm->start_routine)(prm->param);
 689     return 0;
 690 }
 691
 692
 693 int tMPI_Thread_get_hw_number(void)
 694 {
 695     int         ret;
 696
 697     SYSTEM_INFO sysinfo;
 698     GetSystemInfo( &sysinfo );
 699
 700     ret = sysinfo.dwNumberOfProcessors;
 701     return ret;
 702 }
 703
 704
 705
 706
 707 int tMPI_Thread_create(tMPI_Thread_t *thread,
 708                        void *(*start_routine)(void *), void *arg)
 709 {
 710     DWORD thread_id;
 711     struct tMPI_Thread_starter_param *prm;
 712
 713     tMPI_Init_initers();
 714
 715     /* a small memory leak to be sure that it doesn't get deallocated
 716        once this function ends, before the newly created thread uses it. */
 717     prm = (struct tMPI_Thread_starter_param*)
 718         tMPI_Malloc(sizeof(struct tMPI_Thread_starter_param));
 719     prm->start_routine = start_routine;
 720     prm->param         = arg;
 721
 722     *thread = (struct tMPI_Thread*)tMPI_Malloc(sizeof(struct tMPI_Thread)*1);
 723
 724     if (thread == NULL)
 725     {
 726         tMPI_Fatal_error(TMPI_FARGS, "Invalid thread pointer.");
 727         return EINVAL;
 728     }
 729     /* this must be locked before the thread is created to prevent a race
 730        condition if the thread immediately wants to create its own entry */
 731     EnterCriticalSection( &thread_id_list_lock );
 732     /* just create a plain thread. */
 733     (*thread)->started_by_tmpi = 1;
 734     (*thread)->th              = CreateThread(NULL,
 735                                               0,
 736                                               tMPI_Win32_thread_starter,
 737                                               prm,
 738                                               0,
 739                                               &thread_id);
 740     (*thread)->id = thread_id;
 741
 742     if ((*thread)->th == NULL)
 743     {
 744         tMPI_Free(thread);
 745         tMPI_Fatal_error(TMPI_FARGS, "Failed to create thread, error code=%d",
 746                          GetLastError());
 747         return -1;
 748     }
 749     tMPI_Thread_id_list_add_locked(thread_id, (*thread));
 750     LeaveCriticalSection( &thread_id_list_lock );
 751
 752     /* inherit the thread priority from the parent thread. */
 753     /* TODO: is there value in setting this, vs. just allowing it to default
 754        from the process?  currently, this limits the effectivenes of changing
 755        the priority in eg: TaskManager. */
 756     SetThreadPriority(((*thread)->th), GetThreadPriority(GetCurrentThread()));
 757
 758     return 0;
 759 }
 760
 761
 762
 763
 764
 765
 766
 767 int tMPI_Thread_join(tMPI_Thread_t thread, void **value_ptr)
 768 {
 769     DWORD ret, retval;
 770
 771     ret = WaitForSingleObject(thread->th, INFINITE);
 772
 773     if (ret != 0)
 774     {
 775         tMPI_Fatal_error(TMPI_FARGS, "Failed to join thread. error code=%d",
 776                          GetLastError());
 777         return -1;
 778     }
 779
 780     if (value_ptr)
 781     {
 782         if (!GetExitCodeThread(thread, &retval))
 783         {
 784             /* TODO: somehow assign value_ptr */
 785             tMPI_Fatal_error(TMPI_FARGS,
 786                              "Failed to get thread exit code: error=%d",
 787                              GetLastError());
 788             return -1;
 789         }
 790     }
 791     CloseHandle(thread->th);
 792     tMPI_Thread_id_list_remove(thread->id);
 793     tMPI_Free(thread);
 794
 795     return 0;
 796 }
 797
 798
 799 void tMPI_Thread_exit(void *value_ptr)
 800 {
 801     /* TODO: fix exit code */
 802     /* TODO: call destructors for thread-local storage */
 803     ExitThread( 0 );
 804 }
 805
 806
 807
 808
 809 int tMPI_Thread_cancel(tMPI_Thread_t thread)
 810 {
 811     if (!TerminateThread( thread, -1) )
 812     {
 813         tMPI_Fatal_error(TMPI_FARGS, "Failed thread_cancel, error code=%d",
 814                          GetLastError());
 815         return -1;
 816     }
 817     tMPI_Thread_id_list_remove(thread->id);
 818     return 0;
 819 }
 820
 821
 822 tMPI_Thread_t tMPI_Thread_self(void)
 823 {
 824     tMPI_Thread_t th;
 825     tMPI_Init_initers();
 826
 827     th = tMPI_Thread_id_list_add_self();
 828
 829     return th;
 830 }
 831
 832
 833 int tMPI_Thread_equal(tMPI_Thread_t t1, tMPI_Thread_t t2)
 834 {
 835     /* because the tMPI thread IDs are unique, we can compare them directly */
 836     return (t1 == t2);
 837 }
 838
 839 enum tMPI_Thread_setaffinity_support tMPI_Thread_setaffinity_support(void)
 840 {
 841     /* Windows supports seting of thread affinities */
 842     return TMPI_SETAFFINITY_SUPPORT_YES;
 843 }
 844
 845 int tMPI_Thread_setaffinity_single(tMPI_Thread_t thread, unsigned int nr)
 846 {
 847     GROUP_AFFINITY   GroupAffinity;
 848     PROCESSOR_NUMBER IdealProcessorNumber;
 849     /* thread NUMA node */
 850     USHORT           NumaNodeNumber;
 851
 852     /* check for a processor info array. This exists if NUMA
 853        style calls have been succesfully initialized. */
 854     if (g_MPI_ProcessorInfo != NULL)
 855     {
 856
 857         /*func_GetCurrentProcessorNumberEx(&CurrentProcessorNumber);*/
 858         /* group, mask. */
 859         memcpy(&GroupAffinity,
 860                &(g_MPI_ProcessorInfo[nr].GroupAffinity),
 861                sizeof(GROUP_AFFINITY));
 862
 863         /* group, processor number */
 864
 865         memcpy(&IdealProcessorNumber,
 866                &(g_MPI_ProcessorInfo[nr].ProcessorNumber),
 867                sizeof(PROCESSOR_NUMBER));
 868
 869
 870         /* set the NUMA node affinity for the current thread
 871            failures to set the current thread affinity are ignored,
 872            as a fringe case can arise on >32 processor systems with a 32bit
 873            build/code.
 874          */
 875         func_SetThreadIdealProcessorEx(thread->th,
 876                                        &IdealProcessorNumber,
 877                                        NULL);
 878
 879         if (func_GetNumaProcessorNodeEx(&IdealProcessorNumber,
 880                                         &NumaNodeNumber))
 881         {
 882             /* for the NUMA node number associated with the current processor
 883                number, get the group affinity mask */
 884             if (func_GetNumaNodeProcessorMaskEx(NumaNodeNumber,
 885                                                 &GroupAffinity))
 886             {
 887                 /* set the current thread affinity to prevent it from running
 888                    on other NUMA nodes */
 889                 func_SetThreadGroupAffinity(thread->th,
 890                                             &GroupAffinity,
 891                                             NULL);
 892                 return 0;
 893             }
 894         }
 895         return 1;
 896     }
 897     else
 898     {
 899         /* No NUMA-style calls. We just do a simpler thing. */
 900         if ( (func_SetThreadIdealProcessor != NULL) )
 901         {
 902             return (func_SetThreadIdealProcessor(thread->th, nr) == -1);
 903         }
 904     }
 905     return 0;
 906 }
 907
 908
 909
 910 int tMPI_Thread_mutex_init(tMPI_Thread_mutex_t *mtx)
 911 {
 912     if (mtx == NULL)
 913     {
 914         return EINVAL;
 915     }
 916
 917     mtx->mutex = (struct tMPI_Mutex*)tMPI_Malloc(sizeof(struct tMPI_Mutex)*1);
 918     InitializeCriticalSection(&(mtx->mutex->cs));
 919
 920     return 0;
 921 }
 922
 923
 924 int tMPI_Thread_mutex_destroy(tMPI_Thread_mutex_t *mtx)
 925 {
 926     if (mtx == NULL)
 927     {
 928         return EINVAL;
 929     }
 930
 931     DeleteCriticalSection(&(mtx->mutex->cs));
 932     tMPI_Free(mtx->mutex);
 933
 934     return 0;
 935 }
 936
 937
 938
 939
 940 static int tMPI_Thread_mutex_init_once(tMPI_Thread_mutex_t *mtx)
 941 {
 942     int ret = 0;
 943
 944     /* This is essentially a copy of the code from the one-time
 945      * initialization, but with a call to the mutex init routine instead.
 946      * It might seem like overkill, but it will only be executed the first
 947      * time you call a static mutex, and it is important to get all the
 948      * memory barriers right. Trust me, you don't want a deadlock here...
 949      */
 950
 951     /* initialize the initializers */
 952     tMPI_Init_initers();
 953     /* Lock the common one-time init mutex so we can check carefully */
 954     EnterCriticalSection( &mutex_init );
 955
 956     /* Do the actual (locked) check - system mutex is locked if we get here */
 957     if (mtx->mutex == NULL)
 958     {
 959         /* No need to keep the lock during execution -
 960          * Only one thread can do it anyway.
 961          */
 962         ret = tMPI_Thread_mutex_init(mtx);
 963     }
 964     LeaveCriticalSection( &mutex_init );
 965
 966     return ret;
 967 }
 968
 969
 970
 971 int tMPI_Thread_mutex_lock(tMPI_Thread_mutex_t *mtx)
 972 {
 973     /* check whether the mutex is initialized */
 974     if (tMPI_Atomic_get( &(mtx->initialized)  ) == 0)
 975     {
 976         tMPI_Thread_mutex_init_once(mtx);
 977     }
 978
 979     /* The mutex is now guaranteed to be valid. */
 980     EnterCriticalSection( &(mtx->mutex->cs) );
 981
 982     return 0;
 983 }
 984
 985
 986
 987
 988 int tMPI_Thread_mutex_trylock(tMPI_Thread_mutex_t *mtx)
 989 {
 990     BOOL ret;
 991
 992     /* check whether the mutex is initialized */
 993     if (tMPI_Atomic_get( &(mtx->initialized)  ) == 0)
 994     {
 995         tMPI_Thread_mutex_init_once(mtx);
 996     }
 997
 998     /* The mutex is now guaranteed to be valid. */
 999     ret = TryEnterCriticalSection( &(mtx->mutex->cs) );
1000
1001     return (ret != 0);
1002 }
1003
1004
1005
1006 int tMPI_Thread_mutex_unlock(tMPI_Thread_mutex_t *mtx)
1007 {
1008     /* we should have initialized our critical section anyway */
1009     LeaveCriticalSection( &(mtx->mutex->cs) );
1010
1011     return 0;
1012 }
1013
1014
1015
1016 int tMPI_Thread_key_create(tMPI_Thread_key_t *key, void (*destructor)(void *))
1017 {
1018     if (key == NULL)
1019     {
1020         tMPI_Fatal_error(TMPI_FARGS, "Invalid key pointer.");
1021         return EINVAL;
1022     }
1023
1024
1025     /* TODO: make list of destructors for thread-local storage */
1026     key->key = (struct tMPI_Thread_key*)tMPI_Malloc(sizeof(struct
1027                                                            tMPI_Thread_key)*1);
1028
1029     (key)->key->wkey = TlsAlloc();
1030
1031     if ( (key)->key->wkey == TLS_OUT_OF_INDEXES)
1032     {
1033         tMPI_Fatal_error(TMPI_FARGS,
1034                          "Failed to create thread key, error code=%d.",
1035                          GetLastError());
1036         return -1;
1037     }
1038
1039     return 0;
1040 }
1041
1042
1043 int tMPI_Thread_key_delete(tMPI_Thread_key_t key)
1044 {
1045     TlsFree(key.key->wkey);
1046     tMPI_Free(key.key);
1047
1048     return 0;
1049 }
1050
1051
1052
1053 void * tMPI_Thread_getspecific(tMPI_Thread_key_t key)
1054 {
1055     void *p = NULL;
1056
1057     p = TlsGetValue(key.key->wkey);
1058
1059     return p;
1060 }
1061
1062
1063 int tMPI_Thread_setspecific(tMPI_Thread_key_t key, void *value)
1064 {
1065     BOOL ret;
1066
1067     ret = TlsSetValue(key.key->wkey, value);
1068
1069     return ret == 0;
1070 }
1071
1072 #if 0
1073 /* use once Vista is minimum required version */
1074 static BOOL CALLBACK InitHandleWrapperFunction(PINIT_ONCE InitOnce,
1075                                                PVOID      Parameter,
1076                                                PVOID     *lpContext)
1077 {
1078     void (*fn)(void) = (void (*)(void))Parameter;
1079
1080     fn();
1081
1082     return TRUE;
1083 }
1084
1085 CRITICAL_SECTION tMPI_Once_cs;
1086 tMPI_Spinlock_t  tMPI_Once_cs_lock = TMPI_SPINLOCK_INITIALIZER;
1087 volatile int     tMPI_Once_init    = 0;
1088 #endif
1089
1090 int tMPI_Thread_once(tMPI_Thread_once_t *once_control,
1091                      void                (*init_routine)(void))
1092 {
1093 #if 0
1094     /* use once Vista is minimum required version */
1095     BOOL bStatus;
1096     bStatus = InitOnceExecuteOnce(once_control, InitHandleWrapperFunction,
1097                                   init_routine, NULL);
1098
1099     if (!bStatus)
1100     {
1101         tMPI_Fatal_error(TMPI_FARGS, "Failed to run thread_once routine");
1102         return -1;
1103     }
1104 #else
1105     /* really ugly hack - and it's slow... */
1106     tMPI_Init_initers();
1107     EnterCriticalSection(&once_init);
1108     if (tMPI_Atomic_get(&(once_control->once)) == 0)
1109     {
1110         (*init_routine)();
1111         tMPI_Atomic_set(&(once_control->once), 1);
1112     }
1113     LeaveCriticalSection(&once_init);
1114 #endif
1115     return 0;
1116 }
1117
1118
1119
1120
1121
1122 int tMPI_Thread_cond_init(tMPI_Thread_cond_t *cond)
1123 {
1124     if (cond == NULL)
1125     {
1126         return EINVAL;
1127     }
1128
1129     cond->condp = (struct tMPI_Thread_cond*)
1130         tMPI_Malloc(sizeof(struct tMPI_Thread_cond)*1);
1131 #if 0
1132     /* use this code once Vista is the minimum version required */
1133     InitializeConditionVariable( &(cond->cv) );
1134 #else
1135     cond->condp->Nwaiters = 0;
1136     InitializeCriticalSection(&(cond->condp->wtr_lock));
1137     cond->condp->Nrelease = 0;
1138     cond->condp->cycle    = 0;
1139     /* a manual reset, unsignalled event */
1140     cond->condp->ev = CreateEvent(NULL, TRUE, FALSE, NULL);
1141 #endif
1142     return 0;
1143 }
1144
1145
1146 int tMPI_Thread_cond_destroy(tMPI_Thread_cond_t *cond)
1147 {
1148 #if 0
1149     /* use this code once Vista is the minimum version required */
1150     /* windows doesnt have this function */
1151 #else
1152     DeleteCriticalSection(&(cond->condp->wtr_lock));
1153     tMPI_Free(cond->condp);
1154 #endif
1155     return 0;
1156 }
1157
1158
1159
1160 /*! \brief Static init routine for pthread barrier
1161  *
1162  * \internal
1163  *
1164  * This is only used as a wrapper to enable static initialization
1165  * of posix thread types together with out abstraction layer for tMPI_Thread.h
1166  *
1167  * \param cond  Condition variable, must be statically initialized
1168  *
1169  * \return status - 0 on success, or a standard error code.
1170  */
1171 static int tMPI_Thread_cond_init_once(tMPI_Thread_cond_t *cond)
1172 {
1173     int ret = 0;
1174
1175     /* This is essentially a copy of the code from the one-time
1176      * initialization, but with a call to the cond init routine instead.
1177      * It might seem like overkill, but it will only be executed the first
1178      * time you call a static condition variable, and it is important to get
1179      * the memory barriers right. Trust me, you don't want a deadlock here...
1180      */
1181
1182     /* initialize the initializers */
1183     tMPI_Init_initers();
1184     /* Lock the common one-time init mutex so we can check carefully */
1185     EnterCriticalSection( &cond_init );
1186
1187     /* Do the actual (locked) check - system mutex is locked if we get here */
1188     if (cond->condp == NULL)
1189     {
1190         /* No need to keep the lock during execution -
1191          * Only one thread can do it anyway.  */
1192         ret = tMPI_Thread_cond_init(cond);
1193     }
1194     LeaveCriticalSection( &cond_init );
1195
1196     return ret;
1197 }
1198
1199
1200
1201
1202 int tMPI_Thread_cond_wait(tMPI_Thread_cond_t *cond, tMPI_Thread_mutex_t *mtx)
1203 {
1204     BOOL wait_done   = FALSE;
1205     BOOL last_waiter = FALSE;
1206     int  my_cycle;
1207
1208     /* check whether the condition is initialized */
1209     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
1210     {
1211         tMPI_Thread_cond_init_once(cond);
1212     }
1213     /* the mutex must have been initialized because it should be locked here */
1214
1215 #if 0
1216     /* use this code once Vista is the minimum version required */
1217     ret = SleepConditionVariableCS (&(cond->cv), &(mtx->cs), INFINITE);
1218
1219     if (!ret)
1220     {
1221         tMPI_Fatal_error(TMPI_FARGS, "Failed wait for condition, error code=%d",
1222                          GetLastError());
1223         return -1;
1224     }
1225 #else
1226     /* serially increase waiter count */
1227     EnterCriticalSection(&(cond->condp->wtr_lock));
1228     cond->condp->Nwaiters++;
1229     my_cycle = cond->condp->cycle;
1230     LeaveCriticalSection(&(cond->condp->wtr_lock));
1231
1232     /* now it's safe to release the mutex from the fn call */
1233     LeaveCriticalSection(&(mtx->mutex->cs));
1234
1235     /* Loop a wait until we found out we've waited for the right event.
1236        Note that this loop is potentially a busy-wait loop in bad
1237        circumstances (higher priority threads, for example). */
1238     do
1239     {
1240         /* do the actual waiting */
1241         if (WaitForSingleObject( cond->condp->ev, INFINITE ) == WAIT_FAILED)
1242         {
1243             tMPI_Fatal_error(TMPI_FARGS, "Failed event reset, error code=%d",
1244                              GetLastError());
1245             return -1;
1246         }
1247
1248         /* serially check whether we got the right event.  */
1249         EnterCriticalSection(&(cond->condp->wtr_lock));
1250         wait_done = (cond->condp->Nrelease > 0) &&
1251             (cond->condp->cycle != my_cycle);
1252         LeaveCriticalSection(&(cond->condp->wtr_lock));
1253     }
1254     while (!wait_done);
1255
1256     /* We obtain the mutex from the function call */
1257     EnterCriticalSection(&(mtx->mutex->cs));
1258
1259     /* we serially decrease the waiter count and release count */
1260     EnterCriticalSection(&(cond->condp->wtr_lock));
1261     cond->condp->Nwaiters--;
1262     cond->condp->Nrelease--;
1263     last_waiter = (cond->condp->Nrelease == 0);
1264     LeaveCriticalSection(&(cond->condp->wtr_lock));
1265
1266     /* manually release the event if everybody's done with it */
1267     if (last_waiter)
1268     {
1269         if (!ResetEvent( cond->condp->ev ))
1270         {
1271             tMPI_Fatal_error(TMPI_FARGS, "Failed event reset, error code=%d",
1272                              GetLastError());
1273             return -1;
1274         }
1275     }
1276 #endif
1277
1278     return 0;
1279 }
1280
1281
1282
1283
1284 int tMPI_Thread_cond_signal(tMPI_Thread_cond_t *cond)
1285 {
1286     /* check whether the condition is initialized */
1287     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
1288     {
1289         tMPI_Thread_cond_init_once(cond);
1290     }
1291     /* The condition variable is now guaranteed to be valid. */
1292 #if 0
1293     /* use this code once Vista is the minimum version required */
1294     WakeConditionVariable( &(cond->cv) );
1295 #else
1296     EnterCriticalSection(&(cond->condp->wtr_lock));
1297     /* check if we're not still busy with a release. If we are, do nothing. */
1298     if (cond->condp->Nwaiters > cond->condp->Nrelease)
1299     {
1300         cond->condp->Nrelease++;
1301         cond->condp->cycle++;
1302         if (!SetEvent(cond->condp->ev)) /* actually release the
1303                                            waiting threads */
1304         {
1305             tMPI_Fatal_error(TMPI_FARGS, "Failed SetEvent, error code=%d",
1306                              GetLastError());
1307             return -1;
1308         }
1309     }
1310     LeaveCriticalSection(&(cond->condp->wtr_lock));
1311 #endif
1312
1313     return 0;
1314 }
1315
1316
1317
1318 int tMPI_Thread_cond_broadcast(tMPI_Thread_cond_t *cond)
1319 {
1320     /* check whether the condition is initialized */
1321     if (tMPI_Atomic_get( &(cond->initialized)  ) == 0)
1322     {
1323         tMPI_Thread_cond_init_once(cond);
1324     }
1325     /* The condition variable is now guaranteed to be valid. */
1326 #if 0
1327     /* use this code once Vista is the minimum version required */
1328     WakeAllConditionVariable( &(cond->cv) );
1329 #else
1330     EnterCriticalSection(&(cond->condp->wtr_lock));
1331     /* check whether there are any waiters */
1332     if (cond->condp->Nwaiters > 0)
1333     {
1334         cond->condp->Nrelease = cond->condp->Nwaiters;
1335         cond->condp->cycle++;
1336         if (!SetEvent(cond->condp->ev)) /* actually release the
1337                                            waiting threads */
1338         {
1339             tMPI_Fatal_error(TMPI_FARGS, "Failed SetEvent, error code=%d",
1340                              GetLastError());
1341             return -1;
1342         }
1343     }
1344     LeaveCriticalSection(&(cond->condp->wtr_lock));
1345 #endif
1346     return 0;
1347 }
1348
1349
1350
1351
1352 int tMPI_Thread_barrier_init(tMPI_Thread_barrier_t *barrier, int n)
1353 {
1354     if (barrier == NULL)
1355     {
1356         return EINVAL;
1357     }
1358
1359     barrier->barrierp = (struct tMPI_Thread_barrier*)
1360         tMPI_Malloc(sizeof(struct tMPI_Thread_barrier)*1);
1361
1362 #if 0
1363     /* use this once Vista is the oldest supported windows version: */
1364     InitializeCriticalSection(&(barrier->barrierp->cs));
1365     InitializeConditionVariable(&(barrier->barrierp->cv));
1366 #else
1367     tMPI_Thread_mutex_init(&(barrier->barrierp->cs));
1368     tMPI_Thread_cond_init(&(barrier->barrierp->cv));
1369 #endif
1370
1371     barrier->threshold = n;
1372     barrier->count     = n;
1373     barrier->cycle     = 0;
1374
1375     return 0;
1376 }
1377
1378
1379
1380 int tMPI_Thread_barrier_destroy(tMPI_Thread_barrier_t *barrier)
1381 {
1382     if (barrier == NULL)
1383     {
1384         return EINVAL;
1385     }
1386
1387 #if 0
1388     DeleteCriticalSection(&(barrier->barrierp->cs));
1389 #else
1390     tMPI_Thread_mutex_destroy(&(barrier->barrierp->cs));
1391 #endif
1392
1393     tMPI_Thread_cond_destroy(&(barrier->barrierp->cv));
1394
1395     tMPI_Free(barrier->barrierp);
1396
1397     return 0;
1398 }
1399
1400
1401
1402 /*! \brief Static init routine for pthread barrier
1403  *
1404  * \internal
1405  *
1406  * This is only used as a wrapper to enable static initialization
1407  * of posix thread types together with out abstraction layer for tMPI_Thread.h
1408  *
1409  * \param barrier Statically initialized barrier type
1410  * \param n       Number of members in barrier
1411  *
1412  * \return status - 0 on success, or a standard error code.
1413  */
1414 static int tMPI_Thread_barrier_init_once(tMPI_Thread_barrier_t *barrier, int n)
1415 {
1416     int ret;
1417
1418     /* This is essentially a copy of the code from the one-time
1419      * initialization, but with a call to the cond init routine instead.
1420      * It might seem like overkill, but it will only be executed the first
1421      * time you call a static condition variable, and it is important to get
1422      * the memory barriers right. Trust me, you don't want a deadlock here...
1423      */
1424
1425
1426     /* initialize the initializers */
1427     tMPI_Init_initers();
1428
1429     /* Lock the common one-time init mutex so we can check carefully */
1430     EnterCriticalSection( &barrier_init );
1431
1432     /* Do the actual (locked) check - system mutex is locked if we get here */
1433     if (barrier->barrierp == NULL)
1434     {
1435         /* No need to keep the lock during execution -
1436          * Only one thread can do it anyway.  */
1437         ret = tMPI_Thread_barrier_init(barrier, n);
1438     }
1439     LeaveCriticalSection( &barrier_init );
1440
1441     return ret;
1442 }
1443
1444
1445
1446 int tMPI_Thread_barrier_wait(tMPI_Thread_barrier_t *barrier)
1447 {
1448     int     cycle;
1449     BOOL    rc  = FALSE;
1450     int     ret = 0;
1451     /*tMPI_Thread_pthread_barrier_t *p;*/
1452
1453     /* check whether the barrier is initialized */
1454     if (tMPI_Atomic_get( &(barrier->initialized)  ) == 0)
1455     {
1456         tMPI_Thread_barrier_init_once(barrier, barrier->threshold);
1457     }
1458
1459 #if 0
1460     EnterCriticalSection( &(barrier->barrierp->cs)  );
1461 #else
1462     tMPI_Thread_mutex_lock( &(barrier->barrierp->cs) );
1463 #endif
1464
1465
1466
1467     cycle = barrier->cycle;
1468
1469     /* Decrement the count atomically and check if it is zero.
1470      * This will only be true for the last thread calling us.
1471      */
1472     if (--(barrier->count) <= 0)
1473     {
1474         barrier->cycle = !barrier->cycle;
1475         barrier->count = barrier->threshold;
1476 #if 0
1477         WakeAllConditionVariable( &(barrier->barrierp->cv) );
1478 #else
1479         tMPI_Thread_cond_broadcast( &(barrier->barrierp->cv) );
1480 #endif
1481     }
1482     else
1483     {
1484         while (cycle == barrier->cycle)
1485         {
1486 #if 0
1487             rc = SleepConditionVariableCS (&(barrier->barrierp->cv),
1488                                            &(barrier->barrierp->cs),
1489                                            INFINITE);
1490             if (!rc)
1491             {
1492                 ret = -1;
1493                 break;
1494             }
1495 #else
1496             rc = tMPI_Thread_cond_wait(&barrier->barrierp->cv,
1497                                        &barrier->barrierp->cs);
1498             if (rc != 0)
1499             {
1500                 break;
1501             }
1502 #endif
1503         }
1504     }
1505 #if 0
1506     LeaveCriticalSection( &(barrier->barrierp->cs)  );
1507 #else
1508     tMPI_Thread_mutex_unlock( &(barrier->barrierp->cs) );
1509 #endif
1510     return ret;
1511 }
1512
1513 #else
1514
1515 /* just to have some symbols */
1516 int tMPI_Thread_winthreads = 0;
1517
1518 #endif /* THREAD_WINDOWS  */