src/external/thread_mpi/src/tmpi_init.cpp

   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,2018,2021, 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 #ifdef HAVE_TMPI_CONFIG_H
  40 #include "tmpi_config.h"
  41 #endif
  42
  43 #ifdef HAVE_CONFIG_H
  44 #include "config.h"
  45 #endif
  46
  47 #ifdef HAVE_UNISTD_H
  48 #include <unistd.h>
  49 #endif
  50
  51 #include <errno.h>
  52 #include <stdlib.h>
  53 #include <stdio.h>
  54 #include <string.h>
  55 #if !(defined( _WIN32 ) || defined( _WIN64 ) )
  56 #include <sys/time.h>
  57
  58 #endif
  59
  60 #include <cassert>
  61
  62 #include "impl.h"
  63
  64 #ifdef TMPI_TRACE
  65 #include <stdarg.h>
  66 #endif
  67
  68
  69
  70
  71
  72
  73 /* there are a few global variables that maintain information about the
  74    running threads. Some are defined by the MPI standard: */
  75 /* TMPI_COMM_WORLD is in tmpi_malloc.c due to technical reasons */
  76 tMPI_Group TMPI_GROUP_EMPTY = NULL;
  77
  78
  79 /* the threads themselves (tmpi_comm only contains lists of pointers to this
  80       structure */
  81 struct tmpi_thread *threads  = NULL;
  82 int                 Nthreads = 0;
  83
  84 /* thread info */
  85 tMPI_Thread_key_t id_key; /* the key to get the thread id */
  86
  87
  88
  89 /* whether MPI has finalized (we need this to distinguish pre-inited from
  90        post-finalized states */
  91 static tmpi_bool tmpi_finalized = FALSE;
  92
  93 /* misc. global information about MPI */
  94 struct tmpi_global *tmpi_global = NULL;
  95
  96
  97
  98
  99
 100
 101 /* start N threads with argc, argv (used by tMPI_Init)*/
 102 static
 103 int tMPI_Start_threads(tmpi_bool main_returns, int N,
 104                        tMPI_Affinity_strategy aff_strategy,
 105                        int *argc, char ***argv,
 106                        void (*start_fn)(const void*), const void *start_arg,
 107                        int (*start_fn_main)(int, char**));
 108
 109 /* starter function for threads; takes a void pointer to a
 110       struct tmpi_starter_, which calls main() if tmpi_start_.fn == NULL */
 111 static void* tMPI_Thread_starter(void *arg);
 112
 113 /* allocate and initialize the data associated with a thread structure */
 114 static int tMPI_Thread_init(struct tmpi_thread *th);
 115 /* deallocate the data associated with a thread structure */
 116 static void tMPI_Thread_destroy(struct tmpi_thread *th);
 117
 118
 119
 120
 121 #ifdef TMPI_TRACE
 122 void tMPI_Trace_print(const char *fmt, ...)
 123 {
 124     va_list                    argp;
 125     struct tmpi_thread       * th  = NULL;
 126     static tMPI_Thread_mutex_t mtx = TMPI_THREAD_MUTEX_INITIALIZER;
 127
 128     /* don't check for errors during trace */
 129     tMPI_Thread_mutex_lock(&mtx);
 130     if (threads)
 131     {
 132         th = tMPI_Get_current();
 133         printf("THREAD %02d: ", (int)(th-threads));
 134     }
 135     else
 136     {
 137         printf("THREAD main: ");
 138     }
 139     va_start(argp, fmt);
 140     vprintf(fmt, argp);
 141     printf("\n");
 142     fflush(stdout);
 143     va_end(argp);
 144     tMPI_Thread_mutex_unlock(&mtx);
 145 }
 146 #endif
 147
 148
 149 tmpi_bool tMPI_Is_master(void)
 150 {
 151     /* if there are no other threads, we're the main thread */
 152     if ( (!TMPI_COMM_WORLD) || TMPI_COMM_WORLD->grp.N == 0)
 153     {
 154         return TRUE;
 155     }
 156
 157     /* otherwise we know this through thread specific data: */
 158     /* whether the thread pointer points to the head of the threads array */
 159     return (tmpi_bool)(tMPI_Get_current() == threads);
 160 }
 161
 162 tMPI_Comm tMPI_Get_comm_self(void)
 163 {
 164     struct tmpi_thread* th = tMPI_Get_current();
 165     return th->self_comm;
 166 }
 167
 168
 169 int tMPI_Get_N(int *argc, char ***argv, const char *optname, int *nthreads)
 170 {
 171     int i;
 172     int ret = TMPI_SUCCESS;
 173
 174     *nthreads = 0;
 175     if (!optname)
 176     {
 177         i = 0;
 178     }
 179     else
 180     {
 181         for (i = 1; i < *argc; i++)
 182         {
 183             if (strcmp(optname, (*argv)[i]) == 0)
 184             {
 185                 break;
 186             }
 187         }
 188     }
 189     if (i+1 < (*argc))
 190     {
 191         /* the number of processes is an argument */
 192         char *end;
 193         *nthreads = strtol((*argv)[i+1], &end, 10);
 194         if (!end || (*end != 0) )
 195         {
 196             *nthreads = 0;
 197             ret       = TMPI_FAILURE;
 198         }
 199     }
 200     if (*nthreads < 1)
 201     {
 202         int nth = tMPI_Thread_get_hw_number();
 203
 204         if (nth < 1)
 205         {
 206             nth = 1;      /* make sure it's at least 1 */
 207         }
 208         *nthreads = nth;
 209     }
 210
 211     return ret;
 212 }
 213
 214 static int tMPI_Thread_init(struct tmpi_thread *th)
 215 {
 216     int ret;
 217     int N_envelopes      = (Nthreads+1)*N_EV_ALLOC;
 218     int N_send_envelopes = N_EV_ALLOC;
 219     int N_reqs           = (Nthreads+1)*N_EV_ALLOC;
 220     int i;
 221
 222     /* we set our thread id, as a thread-specific piece of global data. */
 223     ret = tMPI_Thread_setspecific(id_key, th);
 224     if (ret != 0)
 225     {
 226         return ret;
 227     }
 228
 229     /* allocate comm.self */
 230     ret = tMPI_Comm_alloc( &(th->self_comm), TMPI_COMM_WORLD, 1);
 231     if (ret != TMPI_SUCCESS)
 232     {
 233         return ret;
 234     }
 235     th->self_comm->grp.peers[0] = th;
 236
 237     /* allocate envelopes */
 238     ret = tMPI_Free_env_list_init( &(th->envelopes), N_envelopes );
 239     if (ret != TMPI_SUCCESS)
 240     {
 241         return ret;
 242     }
 243     /* recv list */
 244     ret = tMPI_Recv_env_list_init( &(th->evr));
 245     if (ret != TMPI_SUCCESS)
 246     {
 247         return ret;
 248     }
 249     /* send lists */
 250     th->evs = (struct send_envelope_list*)tMPI_Malloc(
 251                 sizeof(struct send_envelope_list)*Nthreads);
 252     if (th->evs == NULL)
 253     {
 254         return TMPI_ERR_NO_MEM;
 255     }
 256     for (i = 0; i < Nthreads; i++)
 257     {
 258         ret = tMPI_Send_env_list_init( &(th->evs[i]), N_send_envelopes);
 259         if (ret != TMPI_SUCCESS)
 260         {
 261             return ret;
 262         }
 263     }
 264
 265     tMPI_Atomic_set( &(th->ev_outgoing_received), 0);
 266
 267     tMPI_Event_init( &(th->p2p_event) );
 268
 269     /* allocate requests */
 270     ret = tMPI_Req_list_init(&(th->rql), N_reqs);
 271     if (ret != TMPI_SUCCESS)
 272     {
 273         return ret;
 274     }
 275
 276
 277 #ifdef USE_COLLECTIVE_COPY_BUFFER
 278     /* allcate copy_buffer list */
 279     ret = tMPI_Copy_buffer_list_init(&(th->cbl_multi),
 280                                      (Nthreads+1)*(N_COLL_ENV+1),
 281                                      Nthreads*COPY_BUFFER_SIZE);
 282     if (ret != TMPI_SUCCESS)
 283     {
 284         return ret;
 285     }
 286 #endif
 287
 288 #ifdef TMPI_PROFILE
 289     ret = tMPI_Profile_init(&(th->profile));
 290     if (ret != TMPI_SUCCESS)
 291     {
 292         return ret;
 293     }
 294 #endif
 295     /* now wait for all other threads to come on line, before we
 296        start the MPI program */
 297     ret = tMPI_Thread_barrier_wait( &(tmpi_global->barrier) );
 298     if (ret != 0)
 299     {
 300         return ret;;
 301     }
 302     return ret;
 303 }
 304
 305
 306 static void tMPI_Thread_destroy(struct tmpi_thread *th)
 307 {
 308     int i;
 309
 310     tMPI_Recv_env_list_destroy( &(th->evr));
 311     for (i = 0; i < Nthreads; i++)
 312     {
 313         tMPI_Send_env_list_destroy( &(th->evs[i]));
 314     }
 315     free(th->evs);
 316     tMPI_Free_env_list_destroy( &(th->envelopes) );
 317     tMPI_Event_destroy( &(th->p2p_event) );
 318     tMPI_Req_list_destroy( &(th->rql) );
 319
 320 #ifdef USE_COLLECTIVE_COPY_BUFFER
 321     tMPI_Copy_buffer_list_destroy(&(th->cbl_multi));
 322 #endif
 323
 324     for (i = 0; i < th->argc; i++)
 325     {
 326         free(th->argv[i]);
 327     }
 328 }
 329
 330 static int tMPI_Global_init(struct tmpi_global *g, int Nthreads)
 331 {
 332     int ret;
 333
 334     g->usertypes        = NULL;
 335     g->N_usertypes      = 0;
 336     g->Nalloc_usertypes = 0;
 337     ret                 = tMPI_Thread_mutex_init(&(g->timer_mutex));
 338     if (ret != 0)
 339     {
 340         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 341     }
 342     tMPI_Spinlock_init(&(g->datatype_lock));
 343
 344     ret = tMPI_Thread_barrier_init( &(g->barrier), Nthreads);
 345     if (ret != 0)
 346     {
 347         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 348     }
 349
 350     ret = tMPI_Thread_mutex_init(&(g->comm_link_lock));
 351     if (ret != 0)
 352     {
 353         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 354     }
 355
 356
 357 #if !(defined( _WIN32 ) || defined( _WIN64 ) )
 358     /* the time at initialization. */
 359     gettimeofday( &(g->timer_init), NULL);
 360 #else
 361     /* the time at initialization. */
 362     g->timer_init = GetTickCount();
 363 #endif
 364     return TMPI_SUCCESS;
 365 }
 366
 367 static void tMPI_Global_destroy(struct tmpi_global *g)
 368 {
 369     tMPI_Thread_barrier_destroy(&(g->barrier));
 370     tMPI_Thread_mutex_destroy(&(g->timer_mutex));
 371     tMPI_Thread_mutex_destroy(&(g->comm_link_lock));
 372     for (int i = 0; i < g->N_usertypes; i++)
 373     {
 374         tMPI_Free(g->usertypes[i]->comps);
 375         tMPI_Free(g->usertypes[i]);
 376     }
 377     tMPI_Free(g->usertypes);
 378 }
 379
 380
 381
 382
 383 static void* tMPI_Thread_starter(void *arg)
 384 {
 385     int                 ret;
 386     struct tmpi_thread *th = (struct tmpi_thread*)arg;
 387
 388 #ifdef TMPI_TRACE
 389     tMPI_Trace_print("Created thread nr. %d", (int)(th-threads));
 390 #endif
 391
 392     ret = tMPI_Thread_init(th);
 393     if (ret != TMPI_SUCCESS)
 394     {
 395         return NULL;
 396     }
 397
 398     /* start_fn, start_arg, argc and argv were set by the calling function */
 399     if (!th->start_fn)
 400     {
 401         th->start_fn_main(th->argc, th->argv);
 402     }
 403     else
 404     {
 405         th->start_fn(th->start_arg);
 406         if (!tmpi_finalized)
 407         {
 408             tMPI_Finalize();
 409         }
 410     }
 411
 412     return NULL;
 413 }
 414
 415
 416 int tMPI_Start_threads(tmpi_bool main_returns, int N,
 417                        tMPI_Affinity_strategy aff_strategy,
 418                        int *argc, char ***argv,
 419                        void (*start_fn)(const void*), const void *start_arg,
 420                        int (*start_fn_main)(int, char**))
 421 {
 422     int ret;
 423 #ifdef TMPI_TRACE
 424     tMPI_Trace_print("tMPI_Start_threads(%d, %d, %d, %d, %d, %p, %p, %p, %p)",
 425                      main_returns, N, aff_strategy, argc, argv, start_fn,
 426                      start_arg);
 427 #endif
 428     if (N > 0)
 429     {
 430         int i;
 431         int set_affinity = FALSE;
 432
 433         tmpi_finalized = FALSE;
 434         Nthreads       = N;
 435
 436         /* allocate global data */
 437         tmpi_global = (struct tmpi_global*)
 438             tMPI_Malloc(sizeof(struct tmpi_global));
 439         if (tmpi_global == 0)
 440         {
 441             return TMPI_ERR_NO_MEM;
 442         }
 443         ret = tMPI_Global_init(tmpi_global, N);
 444         if (ret != TMPI_SUCCESS)
 445         {
 446             return ret;
 447         }
 448
 449         /* allocate world and thread data */
 450         threads = (struct tmpi_thread*)
 451             tMPI_Malloc(sizeof(struct tmpi_thread)*N);
 452         if (threads == NULL)
 453         {
 454             return TMPI_ERR_NO_MEM;
 455         }
 456         ret = tMPI_Comm_alloc(&TMPI_COMM_WORLD, NULL, N);
 457         if (ret != TMPI_SUCCESS)
 458         {
 459             return ret;
 460         }
 461         assert(TMPI_COMM_WORLD != nullptr);
 462         TMPI_GROUP_EMPTY = tMPI_Group_alloc();
 463
 464         if (tMPI_Thread_key_create(&id_key, NULL))
 465         {
 466             return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_INIT);
 467         }
 468         for (i = 0; i < N; i++)
 469         {
 470             TMPI_COMM_WORLD->grp.peers[i] = &(threads[i]);
 471
 472             /* copy argc, argv */
 473             if (argc && argv)
 474             {
 475                 int j;
 476                 threads[i].argc = *argc;
 477                 threads[i].argv = (char**)tMPI_Malloc(threads[i].argc*
 478                                                       sizeof(char*));
 479                 for (j = 0; j < threads[i].argc; j++)
 480                 {
 481 #if !(defined( _WIN32 ) || defined( _WIN64 ) )
 482                     threads[i].argv[j] = strdup( (*argv)[j] );
 483 #else
 484                     threads[i].argv[j] = _strdup( (*argv)[j] );
 485 #endif
 486                 }
 487             }
 488             else
 489             {
 490                 threads[i].argc = 0;
 491                 threads[i].argv = NULL;
 492             }
 493             threads[i].start_fn      = start_fn;
 494             threads[i].start_fn_main = start_fn_main;
 495             threads[i].start_arg     = start_arg;
 496         }
 497
 498         /* now check whether to set affinity */
 499         if (aff_strategy == TMPI_AFFINITY_ALL_CORES)
 500         {
 501             int nhw = tMPI_Thread_get_hw_number();
 502             if ((nhw > 1) && (nhw == N))
 503             {
 504                 set_affinity = TRUE;
 505             }
 506         }
 507
 508         /* set thread 0's properties */
 509         threads[0].thread_id = tMPI_Thread_self();
 510         if (set_affinity)
 511         {
 512             /* set the main thread's affinity */
 513             tMPI_Thread_setaffinity_single(threads[0].thread_id, 0);
 514         }
 515
 516         for (i = 1; i < N; i++) /* zero is the main thread */
 517         {
 518             ret = tMPI_Thread_create(&(threads[i].thread_id),
 519                                      tMPI_Thread_starter,
 520                                      (void*)&(threads[i]) );
 521
 522             if (set_affinity)
 523             {
 524                 tMPI_Thread_setaffinity_single(threads[i].thread_id, i);
 525             }
 526             if (ret != TMPI_SUCCESS)
 527             {
 528                 return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_INIT);
 529             }
 530         }
 531         /* the main thread also runs start_fn if we don't want
 532            it to return */
 533         if (!main_returns)
 534         {
 535             tMPI_Thread_starter((void*)&(threads[0]));
 536
 537         }
 538         else
 539         {
 540             ret = tMPI_Thread_init(&(threads[0]));
 541             if (ret != 0)
 542             {
 543                 return ret;
 544             }
 545         }
 546     }
 547     return TMPI_SUCCESS;
 548 }
 549
 550
 551 int tMPI_Init(int *argc, char ***argv,
 552               int (*start_function)(int, char**))
 553 {
 554     int ret;
 555 #ifdef TMPI_TRACE
 556     tMPI_Trace_print("tMPI_Init(%p, %p, %p)", argc, argv, start_function);
 557 #endif
 558
 559     if (TMPI_COMM_WORLD == 0) /* we're the main process */
 560     {
 561         int N = 0;
 562         tMPI_Get_N(argc, argv, "-nt", &N);
 563         ret = tMPI_Start_threads(TRUE, N, TMPI_AFFINITY_ALL_CORES, argc, argv,
 564                                  NULL, NULL, start_function) != 0;
 565         if (ret != 0)
 566         {
 567             return ret;
 568         }
 569     }
 570     else
 571     {
 572         /* if we're a sub-thread we need don't need to do anyhing, because
 573            everything has already been set up by either the main thread,
 574            or the thread runner function.*/
 575     }
 576     return TMPI_SUCCESS;
 577 }
 578
 579
 580
 581
 582 int tMPI_Init_fn(int main_thread_returns, int N,
 583                  tMPI_Affinity_strategy aff_strategy,
 584                  void (*start_function)(const void*), const void *arg)
 585 {
 586     int ret;
 587 #ifdef TMPI_TRACE
 588     tMPI_Trace_print("tMPI_Init_fn(%d, %p, %p)", N, start_function, arg);
 589 #endif
 590
 591     if (N < 1)
 592     {
 593         N = tMPI_Thread_get_hw_number();
 594         if (N < 1)
 595         {
 596             N = 1;    /*because that's what the fn returns if it doesn't know*/
 597         }
 598     }
 599
 600     if (TMPI_COMM_WORLD == 0 && N >= 1) /* we're the main process */
 601     {
 602         ret = tMPI_Start_threads(main_thread_returns, N, aff_strategy,
 603                                  0, 0, start_function, arg, NULL);
 604         if (ret != 0)
 605         {
 606             return ret;
 607         }
 608     }
 609     return TMPI_SUCCESS;
 610 }
 611
 612 int tMPI_Initialized(int *flag)
 613 {
 614 #ifdef TMPI_TRACE
 615     tMPI_Trace_print("tMPI_Initialized(%p)", flag);
 616 #endif
 617
 618     *flag = (TMPI_COMM_WORLD && !tmpi_finalized);
 619
 620     return TMPI_SUCCESS;
 621 }
 622
 623 int tMPI_Finalize(void)
 624 {
 625     int i;
 626     int ret;
 627 #ifdef TMPI_TRACE
 628     tMPI_Trace_print("tMPI_Finalize()");
 629 #endif
 630 #ifdef TMPI_DEBUG
 631     printf("%5d: tMPI_Finalize called\n", tMPI_This_threadnr());
 632     fflush(stdout);
 633 #endif
 634
 635 #ifdef TMPI_PROFILE
 636     {
 637         struct tmpi_thread *cur = tMPI_Get_current();
 638
 639         tMPI_Profile_stop( &(cur->profile) );
 640         ret = tMPI_Thread_barrier_wait( &(tmpi_global->barrier) );
 641         if (ret != 0)
 642         {
 643             return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 644         }
 645
 646         if (tMPI_Is_master())
 647         {
 648             tMPI_Profiles_summarize(Nthreads, threads);
 649         }
 650     }
 651 #endif
 652     ret = tMPI_Thread_barrier_wait( &(tmpi_global->barrier) );
 653     if (ret != 0)
 654     {
 655         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 656     }
 657
 658
 659
 660     if (tMPI_Is_master())
 661     {
 662
 663         /* we just wait for all threads to finish; the order isn't very
 664            relevant, as all threads should arrive at their endpoints soon. */
 665         for (i = 1; i < Nthreads; i++)
 666         {
 667             if (tMPI_Thread_join(threads[i].thread_id, NULL))
 668             {
 669                 return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_FINALIZE);
 670             }
 671             tMPI_Thread_destroy(&(threads[i]));
 672         }
 673         /* at this point, we are the only thread left, so we can
 674            destroy the global structures with impunity. */
 675         tMPI_Thread_destroy(&(threads[0]));
 676         free(threads);
 677
 678         tMPI_Thread_key_delete(id_key);
 679         /* de-allocate all the comm stuctures. */
 680         {
 681             tMPI_Comm cur;
 682
 683             ret = tMPI_Thread_mutex_lock(&(tmpi_global->comm_link_lock));
 684             if (ret != 0)
 685             {
 686                 return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 687             }
 688             cur = TMPI_COMM_WORLD->next;
 689             while (cur && (cur != TMPI_COMM_WORLD) )
 690             {
 691                 tMPI_Comm next = cur->next;
 692                 ret = tMPI_Comm_destroy(cur, FALSE);
 693                 if (ret != 0)
 694                 {
 695                     tMPI_Thread_mutex_unlock(&(tmpi_global->comm_link_lock));
 696                     return ret;
 697                 }
 698                 cur = next;
 699             }
 700             ret = tMPI_Comm_destroy(TMPI_COMM_WORLD, FALSE);
 701             if (ret != 0)
 702             {
 703                 tMPI_Thread_mutex_unlock(&(tmpi_global->comm_link_lock));
 704                 return ret;
 705             }
 706             ret = tMPI_Thread_mutex_unlock(&(tmpi_global->comm_link_lock));
 707             if (ret != 0)
 708             {
 709                 return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_IO);
 710             }
 711
 712         }
 713
 714         tMPI_Group_free(&TMPI_GROUP_EMPTY);
 715         threads          = 0;
 716         TMPI_COMM_WORLD  = NULL;
 717         TMPI_GROUP_EMPTY = NULL;
 718         Nthreads         = 0;
 719
 720         /* deallocate the 'global' structure */
 721         tMPI_Global_destroy(tmpi_global);
 722         free(tmpi_global);
 723
 724         tmpi_finalized = TRUE;
 725     }
 726     else
 727     {
 728         tMPI_Thread_exit(0);
 729     }
 730     return TMPI_SUCCESS;
 731 }
 732
 733
 734 int tMPI_Finalized(int *flag)
 735 {
 736 #ifdef TMPI_TRACE
 737     tMPI_Trace_print("tMPI_Finalized(%p)", flag);
 738 #endif
 739     *flag = tmpi_finalized;
 740
 741     return TMPI_SUCCESS;
 742 }
 743
 744
 745
 746 int tMPI_Abort(tMPI_Comm comm, int errorcode)
 747 {
 748 #ifdef TMPI_TRACE
 749     tMPI_Trace_print("tMPI_Abort(%p, %d)", comm, errorcode);
 750 #endif
 751 #if 0
 752     /* we abort(). This way we can run a debugger on it */
 753     fprintf(stderr, "tMPI_Abort called with error code %d", errorcode);
 754     if (comm == TMPI_COMM_WORLD)
 755     {
 756         fprintf(stderr, " on TMPI_COMM_WORLD");
 757     }
 758     fprintf(stderr, "\n");
 759     fflush(stdout);
 760
 761     abort();
 762 #else
 763     /* we just kill all threads, but not the main process */
 764
 765     if (tMPI_Is_master())
 766     {
 767         if (comm == TMPI_COMM_WORLD)
 768         {
 769             fprintf(stderr,
 770                     "tMPI_Abort called on TMPI_COMM_WORLD main with errorcode=%d\n",
 771                     errorcode);
 772         }
 773         else
 774         {
 775             fprintf(stderr,
 776                     "tMPI_Abort called on main thread with errorcode=%d\n",
 777                     errorcode);
 778         }
 779         fflush(stderr);
 780         exit(errorcode);
 781     }
 782     else
 783     {
 784         int *ret;
 785         /* kill myself */
 786         fprintf(stderr, "tMPI_Abort called with error code %d on thread %d\n",
 787                 errorcode, tMPI_This_threadnr());
 788         fflush(stderr);
 789         ret = (int*)malloc(sizeof(int));
 790         tMPI_Thread_exit(ret);
 791     }
 792 #endif
 793     return TMPI_SUCCESS;
 794 }
 795
 796
 797 int tMPI_Get_processor_name(char *name, int *resultlen)
 798 {
 799     int                nr     = tMPI_Threadnr(tMPI_Get_current());
 800     unsigned int       digits = 0;
 801     const unsigned int base   = 10;
 802
 803 #ifdef TMPI_TRACE
 804     tMPI_Trace_print("tMPI_Get_processor_name(%p, %p)", name, resultlen);
 805 #endif
 806     /* we don't want to call sprintf here (it turns out to be not entirely
 807        thread-safe on Mac OS X, for example), so we do it our own way: */
 808
 809     /* first determine number of digits */
 810     {
 811         int rest = nr;
 812         while (rest > 0)
 813         {
 814             rest /= base;
 815             digits++;
 816         }
 817         if (digits == 0)
 818         {
 819             digits = 1;
 820         }
 821     }
 822 #ifndef _MSC_VER
 823     strcpy(name, "thread #");
 824 #else
 825     strncpy_s(name, TMPI_MAX_PROCESSOR_NAME, "thread #", TMPI_MAX_PROCESSOR_NAME);
 826 #endif
 827     /* now construct the number */
 828     {
 829         size_t       len = strlen(name);
 830         unsigned int i;
 831         int          rest = nr;
 832
 833         for (i = 0; i < digits; i++)
 834         {
 835             size_t pos = len + (digits-i-1);
 836             if (pos < (TMPI_MAX_PROCESSOR_NAME -1) )
 837             {
 838                 name[ pos ] = (char)('0' + rest%base);
 839             }
 840             rest /= base;
 841         }
 842         if ( (digits+len) < TMPI_MAX_PROCESSOR_NAME)
 843         {
 844             name[digits + len] = '\0';
 845         }
 846         else
 847         {
 848             name[TMPI_MAX_PROCESSOR_NAME] = '\0';
 849         }
 850
 851     }
 852     if (resultlen)
 853     {
 854         *resultlen = (int)strlen(name); /* For some reason the MPI standard
 855                                            uses ints instead of size_ts for
 856                                            sizes. */
 857     }
 858     return TMPI_SUCCESS;
 859 }
 860
 861
 862
 863
 864
 865 /* TODO: there must be better ways to do this */
 866 double tMPI_Wtime(void)
 867 {
 868     double ret = 0;
 869
 870 #ifdef TMPI_TRACE
 871     tMPI_Trace_print("tMPI_Wtime()");
 872 #endif
 873
 874 #if !(defined( _WIN32 ) || defined( _WIN64 ) )
 875     {
 876         struct timeval tv;
 877         long int       secdiff;
 878         int            usecdiff;
 879
 880         gettimeofday(&tv, NULL);
 881         secdiff  = tv.tv_sec - tmpi_global->timer_init.tv_sec;
 882         usecdiff = tv.tv_usec - tmpi_global->timer_init.tv_usec;
 883
 884         ret = (double)secdiff + 1e-6*usecdiff;
 885     }
 886 #else
 887     {
 888         DWORD tv = GetTickCount();
 889
 890         /* the windows absolute time GetTickCount() wraps around in ~49 days,
 891            so it's safer to always use differences, and assume that our
 892            program doesn't run that long.. */
 893         ret = 1e-3*((unsigned int)(tv - tmpi_global->timer_init));
 894     }
 895 #endif
 896     return ret;
 897 }
 898
 899 double tMPI_Wtick(void)
 900 {
 901 #if !(defined( _WIN32 ) || defined( _WIN64 ) )
 902     /* In Unix, we don't really know. Any modern OS should be at least
 903        this precise, though */
 904     return 1./100.;
 905 #else
 906     /* According to the Windows documentation, this is about right: */
 907     return 1./100.;
 908 #endif
 909 }
 910
 911 int tMPI_Get_count(tMPI_Status *status, tMPI_Datatype datatype, int *count)
 912 {
 913 #ifdef TMPI_TRACE
 914     tMPI_Trace_print("tMPI_Get_count(%p, %p, %p)", status, datatype, count);
 915 #endif
 916     if (!status)
 917     {
 918         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_STATUS);
 919     }
 920     *count = (int)(status->transferred/datatype->size);
 921     return TMPI_SUCCESS;
 922 }