src/gromacs/gmxlib/thread_mpi/p2p_wait.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 #ifdef HAVE_TMPI_CONFIG_H
  39 #include "tmpi_config.h"
  40 #endif
  41
  42 #ifdef HAVE_CONFIG_H
  43 #include "config.h"
  44 #endif
  45
  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 <stdarg.h>
  55 #include <string.h>
  56
  57
  58 #include "impl.h"
  59 #include "p2p.h"
  60
  61
  62 int tMPI_Wait(tMPI_Request *request, tMPI_Status *status)
  63 {
  64     int                 ret = TMPI_SUCCESS;
  65     struct tmpi_thread *cur = tMPI_Get_current();
  66     struct req_list    *rql = &(cur->rql);
  67     struct tmpi_req_   *rq;
  68
  69 #ifdef TMPI_PROFILE
  70     tMPI_Profile_count_start(cur);
  71 #endif
  72 #ifdef TMPI_TRACE
  73     tMPI_Trace_print("tMPI_Wait(%p, %p)", request, status);
  74 #endif
  75     if (!request || !(*request))
  76     {
  77         return TMPI_SUCCESS;
  78     }
  79
  80     rq = *request;
  81     /* fix the pointers */
  82     rq->next = rq;
  83     rq->prev = rq;
  84
  85     /* and wait for our request */
  86     do
  87     {
  88         if (tMPI_Test_single(cur, rq))
  89         {
  90             break;
  91         }
  92         tMPI_Wait_process_incoming(cur);
  93     }
  94     while (TRUE);
  95
  96     rq->ev = NULL; /* we won't be using that envelope any more */
  97     ret    = rq->error;
  98
  99     tMPI_Set_status(rq, status);
 100
 101     /* deallocate */
 102     tMPI_Return_req(rql, *request);
 103
 104 #ifdef TMPI_PROFILE
 105     tMPI_Profile_count_stop(cur, TMPIFN_Wait);
 106 #endif
 107     return ret;
 108 }
 109
 110 int tMPI_Test(tMPI_Request *request, int *flag, tMPI_Status *status)
 111 {
 112     int                 ret = TMPI_SUCCESS;
 113     struct tmpi_thread *cur = tMPI_Get_current();
 114     struct req_list    *rql = &(cur->rql);
 115     struct tmpi_req_   *rq;
 116
 117 #ifdef TMPI_PROFILE
 118     tMPI_Profile_count_start(cur);
 119 #endif
 120 #ifdef TMPI_TRACE
 121     tMPI_Trace_print("tMPI_Test(%p, %p, %p)", request, flag, status);
 122 #endif
 123     if (!request || !(*request))
 124     {
 125         return TMPI_SUCCESS;
 126     }
 127
 128     rq = *request;
 129     /* fix the pointers */
 130     rq->next = rq;
 131     rq->prev = rq;
 132
 133     /* and check our request */
 134     if (tMPI_Test_single(cur, rq))
 135     {
 136         *flag = TRUE;
 137     }
 138
 139     ret = rq->error;
 140
 141     if (rq->finished)
 142     {
 143         tMPI_Set_status(rq, status);
 144         /* deallocate */
 145         tMPI_Return_req(rql, *request);
 146         *request = TMPI_REQUEST_NULL;
 147     }
 148
 149 #ifdef TMPI_PROFILE
 150     tMPI_Profile_count_stop(cur, TMPIFN_Test);
 151 #endif
 152     return ret;
 153 }
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170 /* test multiple requests by first making a linked list of them, and then
 171    checking for their completion. Used in tMPI_{Test|Wait}{all|any|some}
 172
 173    wait = whether to wait for incoming events
 174    blocking = whether to block until all reqs are completed */
 175 static void tMPI_Test_multi_req(struct tmpi_thread *cur,
 176                                 int count, tMPI_Request *array_of_requests,
 177                                 tmpi_bool wait, tmpi_bool blocking)
 178 {
 179     int               i;
 180     struct tmpi_req_ *first = NULL, *last = NULL;
 181
 182     /* construct the list of requests */
 183     for (i = 0; i < count; i++)
 184     {
 185         struct tmpi_req_ *curr = array_of_requests[i];
 186         if (curr)
 187         {
 188             if (!first)
 189             {
 190                 first = curr;
 191             }
 192             /* fix the pointers */
 193             if (!last)
 194             {
 195                 /* we connect to itself */
 196                 last       = curr;
 197                 last->next = NULL;
 198                 last->prev = NULL;
 199             }
 200             else
 201             {
 202                 /* we connect to the last */
 203                 curr->next = NULL;
 204                 curr->prev = last;
 205                 last->next = curr;
 206                 last       = curr;
 207             }
 208         }
 209     }
 210     /* and wait for our request */
 211     do
 212     {
 213         if (tMPI_Test_multi(cur, first, NULL))
 214         {
 215             break;
 216         }
 217         if (wait)
 218         {
 219             tMPI_Wait_process_incoming(cur);
 220         }
 221     }
 222     while (blocking && wait);
 223 }
 224
 225
 226
 227 int tMPI_Waitall(int count, tMPI_Request *array_of_requests,
 228                  tMPI_Status *array_of_statuses)
 229 {
 230     int                 i;
 231     int                 ret = TMPI_SUCCESS;
 232     struct tmpi_thread *cur = tMPI_Get_current();
 233     struct req_list    *rql = &(cur->rql);
 234 #ifdef TMPI_PROFILE
 235     tMPI_Profile_count_start(cur);
 236 #endif
 237 #ifdef TMPI_TRACE
 238     tMPI_Trace_print("tMPI_Waitall(%d, %p, %p)", count, array_of_requests,
 239                      array_of_statuses);
 240 #endif
 241     tMPI_Test_multi_req(cur, count, array_of_requests, TRUE, TRUE);
 242
 243     /* deallocate the now finished requests */
 244     for (i = 0; i < count; i++)
 245     {
 246         if (array_of_requests[i])
 247         {
 248             if (array_of_statuses)
 249             {
 250                 tMPI_Set_status(array_of_requests[i], &(array_of_statuses[i]));
 251             }
 252             if (array_of_requests[i]->error != TMPI_SUCCESS)
 253             {
 254                 ret = TMPI_ERR_IN_STATUS;
 255             }
 256             tMPI_Return_req(rql, array_of_requests[i]);
 257             array_of_requests[i] = TMPI_REQUEST_NULL;
 258         }
 259     }
 260
 261 #ifdef TMPI_PROFILE
 262     tMPI_Profile_count_stop(cur, TMPIFN_Waitall);
 263 #endif
 264     return ret;
 265 }
 266
 267 int tMPI_Testall(int count, tMPI_Request *array_of_requests,
 268                  int *flag, tMPI_Status *array_of_statuses)
 269 {
 270     int                 i;
 271     int                 ret = TMPI_SUCCESS;
 272     struct tmpi_thread *cur = tMPI_Get_current();
 273     struct req_list    *rql = &(cur->rql);
 274 #ifdef TMPI_PROFILE
 275     tMPI_Profile_count_start(cur);
 276 #endif
 277 #ifdef TMPI_TRACE
 278     tMPI_Trace_print("tMPI_Testall(%d, %p, %p, %p)", count, array_of_requests,
 279                      flag, array_of_statuses);
 280 #endif
 281     tMPI_Test_multi_req(cur, count, array_of_requests, FALSE, TRUE);
 282
 283     if (flag)
 284     {
 285         *flag = 1;
 286     }
 287     /* deallocate the possibly finished requests */
 288     for (i = 0; i < count; i++)
 289     {
 290         if (array_of_requests[i] && array_of_requests[i]->finished)
 291
 292         {
 293             if (array_of_statuses)
 294             {
 295                 tMPI_Set_status(array_of_requests[i], &(array_of_statuses[i]));
 296             }
 297             if (array_of_requests[i]->error != TMPI_SUCCESS)
 298             {
 299                 ret = TMPI_ERR_IN_STATUS;
 300             }
 301             tMPI_Return_req(rql, array_of_requests[i]);
 302             array_of_requests[i] = TMPI_REQUEST_NULL;
 303         }
 304         else
 305         {
 306             if (flag)
 307             {
 308                 *flag = 0;
 309             }
 310         }
 311
 312     }
 313
 314 #ifdef TMPI_PROFILE
 315     tMPI_Profile_count_stop(cur, TMPIFN_Testall);
 316 #endif
 317     return ret;
 318 }
 319
 320
 321 int tMPI_Waitany(int count, tMPI_Request *array_of_requests, int *index,
 322                  tMPI_Status *status)
 323 {
 324     int                 i;
 325     int                 ret = TMPI_SUCCESS;
 326     struct tmpi_thread *cur = tMPI_Get_current();
 327     struct req_list    *rql = &(cur->rql);
 328 #ifdef TMPI_PROFILE
 329     tMPI_Profile_count_start(cur);
 330 #endif
 331 #ifdef TMPI_TRACE
 332     tMPI_Trace_print("tMPI_Waitany(%d, %p, %p, %p)", count, array_of_requests,
 333                      index, status);
 334 #endif
 335
 336     tMPI_Test_multi_req(cur, count, array_of_requests, TRUE, FALSE);
 337
 338     /* deallocate the possibly finished requests */
 339     for (i = 0; i < count; i++)
 340     {
 341         if (array_of_requests[i] && array_of_requests[i]->finished)
 342         {
 343             tMPI_Set_status(array_of_requests[i], status);
 344             if (index)
 345             {
 346                 *index = i;
 347             }
 348             if (array_of_requests[i]->error != TMPI_SUCCESS)
 349             {
 350                 ret = TMPI_ERR_IN_STATUS;
 351             }
 352             tMPI_Return_req(rql, array_of_requests[i]);
 353             array_of_requests[i] = TMPI_REQUEST_NULL;
 354             /* and we only need one */
 355             break;
 356         }
 357     }
 358
 359 #ifdef TMPI_PROFILE
 360     tMPI_Profile_count_stop(cur, TMPIFN_Waitany);
 361 #endif
 362     return ret;
 363 }
 364
 365
 366
 367 int tMPI_Testany(int count, tMPI_Request *array_of_requests, int *index,
 368                  int *flag, tMPI_Status *status)
 369 {
 370     int                 i;
 371     int                 ret = TMPI_SUCCESS;
 372     struct tmpi_thread *cur = tMPI_Get_current();
 373     struct req_list    *rql = &(cur->rql);
 374 #ifdef TMPI_PROFILE
 375     tMPI_Profile_count_start(cur);
 376 #endif
 377 #ifdef TMPI_TRACE
 378     tMPI_Trace_print("tMPI_Testany(%d, %p, %p %p, %p)", count,
 379                      array_of_requests, flag, index, status);
 380 #endif
 381
 382     tMPI_Test_multi_req(cur, count, array_of_requests, FALSE, FALSE);
 383
 384     if (flag)
 385     {
 386         *flag = 0;
 387     }
 388     if (index)
 389     {
 390         *index = TMPI_UNDEFINED;
 391     }
 392     /* deallocate the possibly finished requests */
 393     for (i = 0; i < count; i++)
 394     {
 395         if (array_of_requests[i] && array_of_requests[i]->finished)
 396         {
 397             tMPI_Set_status(array_of_requests[i], status);
 398             if (index)
 399             {
 400                 *index = i;
 401             }
 402             if (flag)
 403             {
 404                 *flag = 1;
 405             }
 406             if (array_of_requests[i]->error != TMPI_SUCCESS)
 407             {
 408                 ret = TMPI_ERR_IN_STATUS;
 409             }
 410             tMPI_Return_req(rql, array_of_requests[i]);
 411             array_of_requests[i] = TMPI_REQUEST_NULL;
 412             /* and we only need one */
 413             break;
 414         }
 415     }
 416
 417 #ifdef TMPI_PROFILE
 418     tMPI_Profile_count_stop(cur, TMPIFN_Testany);
 419 #endif
 420     return ret;
 421 }
 422
 423
 424
 425 int tMPI_Waitsome(int incount, tMPI_Request *array_of_requests,
 426                   int *outcount, int *array_of_indices,
 427                   tMPI_Status *array_of_statuses)
 428 {
 429     int                 i;
 430     int                 ret = TMPI_SUCCESS;
 431     struct tmpi_thread *cur = tMPI_Get_current();
 432     struct req_list    *rql = &(cur->rql);
 433 #ifdef TMPI_PROFILE
 434     tMPI_Profile_count_start(cur);
 435 #endif
 436 #ifdef TMPI_TRACE
 437     tMPI_Trace_print("tMPI_Waitsome(%d, %p, %p, %p, %p)", incount,
 438                      array_of_requests, outcount, array_of_indices,
 439                      array_of_statuses);
 440 #endif
 441     tMPI_Test_multi_req(cur, incount, array_of_requests, TRUE, FALSE);
 442
 443     (*outcount) = 0;
 444     /* deallocate the possibly finished requests */
 445     for (i = 0; i < incount; i++)
 446     {
 447         if (array_of_requests[i] && array_of_requests[i]->finished)
 448         {
 449             array_of_indices[*outcount]++;
 450             (*outcount)++;
 451             if (array_of_statuses)
 452             {
 453                 tMPI_Set_status(array_of_requests[i], &(array_of_statuses[i]));
 454             }
 455             if (array_of_requests[i]->error != TMPI_SUCCESS)
 456             {
 457                 ret = TMPI_ERR_IN_STATUS;
 458             }
 459             tMPI_Return_req(rql, array_of_requests[i]);
 460             array_of_requests[i] = TMPI_REQUEST_NULL;
 461         }
 462     }
 463
 464 #ifdef TMPI_PROFILE
 465     tMPI_Profile_count_stop(cur, TMPIFN_Waitsome);
 466 #endif
 467     return ret;
 468 }
 469
 470 int tMPI_Testsome(int incount, tMPI_Request *array_of_requests,
 471                   int *outcount, int *array_of_indices,
 472                   tMPI_Status *array_of_statuses)
 473 {
 474     int                 i;
 475     int                 ret = TMPI_SUCCESS;
 476     struct tmpi_thread *cur = tMPI_Get_current();
 477     struct req_list    *rql = &(cur->rql);
 478 #ifdef TMPI_PROFILE
 479     tMPI_Profile_count_start(cur);
 480 #endif
 481 #ifdef TMPI_TRACE
 482     tMPI_Trace_print("tMPI_Testsome(%d, %p, %p, %p, %p)", incount,
 483                      array_of_requests, outcount, array_of_indices,
 484                      array_of_statuses);
 485 #endif
 486     tMPI_Test_multi_req(cur, incount, array_of_requests, FALSE, TRUE);
 487
 488     (*outcount) = 0;
 489     /* deallocate the possibly finished requests */
 490     for (i = 0; i < incount; i++)
 491     {
 492         if (array_of_requests[i] && array_of_requests[i]->finished)
 493         {
 494             array_of_indices[*outcount]++;
 495             (*outcount)++;
 496             if (array_of_statuses)
 497             {
 498                 tMPI_Set_status(array_of_requests[i], &(array_of_statuses[i]));
 499             }
 500             if (array_of_requests[i]->error != TMPI_SUCCESS)
 501             {
 502                 ret = TMPI_ERR_IN_STATUS;
 503             }
 504             tMPI_Return_req(rql, array_of_requests[i]);
 505             array_of_requests[i] = TMPI_REQUEST_NULL;
 506         }
 507     }
 508
 509
 510 #ifdef TMPI_PROFILE
 511     tMPI_Profile_count_stop(cur, TMPIFN_Testsome);
 512 #endif
 513     return ret;
 514 }