src/gmxlib/thread_mpi/p2p_send_recv.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 /* point-to-point communication exported functions */
  63
  64 int tMPI_Send(void* buf, int count, tMPI_Datatype datatype, int dest,
  65               int tag, tMPI_Comm comm)
  66 {
  67     struct envelope *sev;
  68     struct tmpi_thread *send_dst=tMPI_Get_thread(comm, dest);
  69     struct tmpi_thread *cur=tMPI_Get_current();
  70     struct tmpi_req_ req;
  71
  72 #ifdef TMPI_PROFILE
  73     tMPI_Profile_count_start(cur);
  74 #endif
  75 #ifdef TMPI_TRACE
  76     tMPI_Trace_print("tMPI_Send(%p, %d, %p, %d, %d, %p)", buf, count,
  77                        datatype, dest, tag, comm);
  78 #endif
  79     if (!comm)
  80     {
  81         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_COMM);
  82     }
  83     if (!send_dst)
  84     {
  85         return tMPI_Error(comm, TMPI_ERR_SEND_DEST);
  86     }
  87
  88     sev=tMPI_Post_send(cur, comm, send_dst, buf, count, datatype, tag, FALSE);
  89     tMPI_Req_init(&req, sev);
  90     tMPI_Wait_single(cur, &req);
  91
  92 #ifdef TMPI_PROFILE
  93     tMPI_Profile_count_stop(cur,TMPIFN_Send);
  94 #endif
  95     return req.error;
  96 }
  97
  98
  99
 100
 101 int tMPI_Recv(void* buf, int count, tMPI_Datatype datatype, int source,
 102              int tag, tMPI_Comm comm, tMPI_Status *status)
 103 {
 104     struct envelope *rev;
 105     struct tmpi_thread *recv_src=0;
 106     struct tmpi_thread *cur=tMPI_Get_current();
 107     struct tmpi_req_ req;
 108
 109 #ifdef TMPI_PROFILE
 110     tMPI_Profile_count_start(cur);
 111 #endif
 112 #ifdef TMPI_TRACE
 113     tMPI_Trace_print("tMPI_Recv(%p, %d, %p, %d, %d, %p, %p)", buf, count,
 114                        datatype, source, tag, comm, status);
 115 #endif
 116     if (!comm)
 117     {
 118         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_COMM);
 119     }
 120
 121     if (source!=TMPI_ANY_SOURCE)
 122     {
 123         recv_src=tMPI_Get_thread(comm, source);
 124         if (!recv_src)
 125         {
 126             return tMPI_Error(comm, TMPI_ERR_RECV_SRC);
 127         }
 128     }
 129
 130     rev=tMPI_Post_match_recv(cur, comm, recv_src, buf, count, datatype, tag,
 131                             FALSE);
 132     tMPI_Req_init(&req, rev);
 133     tMPI_Wait_single(cur, &req);
 134
 135     tMPI_Set_status(&req, status);
 136 #ifdef TMPI_PROFILE
 137     tMPI_Profile_count_stop(cur, TMPIFN_Recv);
 138 #endif
 139     return req.error;
 140 }
 141
 142
 143
 144
 145
 146 int tMPI_Sendrecv(void *sendbuf, int sendcount, tMPI_Datatype sendtype,
 147                   int dest, int sendtag, void *recvbuf, int recvcount,
 148                   tMPI_Datatype recvtype, int source, int recvtag,
 149                   tMPI_Comm comm, tMPI_Status *status)
 150 {
 151     struct envelope *rev, *sev;
 152     struct tmpi_thread *cur=tMPI_Get_current();
 153     struct tmpi_thread *recv_src=0;
 154     struct tmpi_thread *send_dst=tMPI_Get_thread(comm, dest);
 155     struct tmpi_req_ sreq, rreq;
 156     int ret=TMPI_SUCCESS;
 157
 158 #ifdef TMPI_PROFILE
 159     tMPI_Profile_count_start(cur);
 160 #endif
 161 #ifdef TMPI_TRACE
 162     tMPI_Trace_print("tMPI_Sendrecv(%p, %d, %p, %d, %d, %p, %d, %p, %d, %d, %p, %p)",
 163                        sendbuf, sendcount, sendtype, dest, sendtag, recvbuf,
 164                        recvcount, recvtype, source, recvtag, comm, status);
 165 #endif
 166     if (!comm)
 167     {
 168         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_COMM);
 169     }
 170     if (!send_dst)
 171     {
 172         return tMPI_Error(comm, TMPI_ERR_SEND_DEST);
 173     }
 174     if (source!=TMPI_ANY_SOURCE)
 175     {
 176         recv_src=tMPI_Get_thread(comm, source);
 177         if (!recv_src)
 178         {
 179             return tMPI_Error(comm, TMPI_ERR_RECV_SRC);
 180         }
 181     }
 182
 183    /* we first prepare to send */
 184     sev=tMPI_Post_send(cur, comm, send_dst, sendbuf, sendcount,
 185                       sendtype, sendtag, FALSE);
 186     tMPI_Req_init(&sreq, sev);
 187     /* the we prepare to receive */
 188     rev=tMPI_Post_match_recv(cur, comm, recv_src, recvbuf, recvcount,
 189                             recvtype, recvtag, FALSE);
 190     tMPI_Req_init(&rreq, rev);
 191
 192     /* fix the pointers */
 193     sreq.next=&rreq;
 194     sreq.prev=NULL;
 195     rreq.prev=&sreq;
 196     rreq.next=NULL;
 197
 198     /* and wait for our requests */
 199     do
 200     {
 201         if (tMPI_Test_multi(cur, &sreq, NULL))
 202             break;
 203         tMPI_Wait_process_incoming(cur);
 204     } while(TRUE);
 205
 206 #ifdef TMPI_PROFILE
 207     tMPI_Profile_count_stop(cur, TMPIFN_Sendrecv);
 208 #endif
 209
 210     tMPI_Set_status(&rreq, status);
 211     ret=sreq.error;
 212     if (rreq.error != TMPI_SUCCESS)
 213         ret=rreq.error;
 214
 215     return ret;
 216 }
 217
 218
 219
 220
 221
 222 /* async */
 223
 224 int tMPI_Isend(void* buf, int count, tMPI_Datatype datatype, int dest,
 225                int tag, tMPI_Comm comm, tMPI_Request *request)
 226 {
 227     struct tmpi_thread *cur=tMPI_Get_current();
 228     struct req_list *rql=&(cur->rql);
 229     struct tmpi_req_ *rq=tMPI_Get_req(rql);
 230     struct tmpi_thread *send_dst=tMPI_Get_thread(comm, dest);
 231     struct envelope *ev;
 232
 233 #ifdef TMPI_PROFILE
 234     tMPI_Profile_count_start(cur);
 235 #endif
 236 #ifdef TMPI_TRACE
 237     tMPI_Trace_print("tMPI_Isend(%p, %d, %p, %d, %d, %p, %p)", buf, count,
 238                        datatype, dest, tag, comm, request);
 239 #endif
 240     if (!comm)
 241     {
 242         tMPI_Return_req(rql,rq);
 243         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_COMM);
 244     }
 245     if (!send_dst)
 246     {
 247         tMPI_Return_req(rql,rq);
 248         return tMPI_Error(comm, TMPI_ERR_SEND_DEST);
 249     }
 250     ev=tMPI_Post_send(cur, comm, send_dst, buf, count, datatype, tag, TRUE);
 251     tMPI_Req_init(rq, ev);
 252     *request=rq;
 253
 254 #ifdef TMPI_PROFILE
 255     tMPI_Profile_count_stop(cur, TMPIFN_Isend);
 256 #endif
 257     return ev->error;
 258 }
 259
 260
 261 int tMPI_Irecv(void* buf, int count, tMPI_Datatype datatype, int source,
 262                int tag, tMPI_Comm comm, tMPI_Request *request)
 263 {
 264     struct tmpi_thread *cur=tMPI_Get_current();
 265     struct req_list *rql=&(cur->rql);
 266     struct tmpi_req_ *rq=tMPI_Get_req(rql);
 267     struct tmpi_thread *recv_src=0;
 268     struct envelope *ev;
 269
 270 #ifdef TMPI_PROFILE
 271     tMPI_Profile_count_start(cur);
 272 #endif
 273 #ifdef TMPI_TRACE
 274     tMPI_Trace_print("tMPI_Irecv(%p, %d, %p, %d, %d, %p, %p)", buf, count,
 275                        datatype, source, tag, comm, request);
 276 #endif
 277     if (!comm)
 278     {
 279         tMPI_Return_req(rql,rq);
 280         return tMPI_Error(TMPI_COMM_WORLD, TMPI_ERR_COMM);
 281     }
 282
 283     if (source!=TMPI_ANY_SOURCE)
 284     {
 285         recv_src=tMPI_Get_thread(comm, source);
 286         if (!recv_src)
 287         {
 288             tMPI_Return_req(rql,rq);
 289             return tMPI_Error(comm, TMPI_ERR_RECV_SRC);
 290         }
 291     }
 292     ev=tMPI_Post_match_recv(cur, comm, recv_src, buf, count, datatype, tag,
 293                             TRUE);
 294     tMPI_Req_init(rq, ev);
 295     *request=rq;
 296 #ifdef TMPI_PROFILE
 297     tMPI_Profile_count_stop(cur, TMPIFN_Irecv);
 298 #endif
 299     return ev->error;
 300 }
 301
 302
 303
 304