/[gcl]/gcl/pargcl/src/mpinu/recv-cache.c
ViewVC logotype

Diff of /gcl/pargcl/src/mpinu/recv-cache.c

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.1 by gene, Fri Oct 28 23:00:29 2005 UTC revision 1.2 by gene, Fri Dec 2 07:54:27 2005 UTC
# Line 37  Line 37 
37  #include "mpi.h"  #include "mpi.h"
38  #include "mpiimpl.h"  #include "mpiimpl.h"
39    
40    typedef enum { BUF_TAG_EQUAL, BUF_TAG_NOT_EQUAL } buf_tag_equality_t;
41    
42  static void buf_init ();  static void buf_init ();
43  static void buf_reset(int rank);  static buf_tag_equality_t buf_tag_compare(int rank, int tag);
44  static int buf_avail (int rank, int len);  static void buf_reset(int source, int tag);
45    static int buf_avail (int source, int len);
46    static int buf_avail_tag(int source, int len, int tag);
47  static void buf_enqueue(int rank, void *buf, int len);  static void buf_enqueue(int rank, void *buf, int len);
48  static void buf_dequeue(int rank, void *buf, int len, int peek_msg);  static void buf_dequeue(int rank, void *buf, int len, int peek_msg);
49  static void buf_peek(int rank, void *buf, int len);  static void *buf_peek(int rank, int len);
50  static void buf_skip(int rank, int len);  static void buf_skip(int rank, int len);
51    
52  /* non-blocking, returns socket descriptor or -1 if none available */  /* non-blocking, returns socket descriptor or -1 if none available
53  int MPINU_rank_of_msg_avail_in_cache(int source) {   * NOTE:  This function assumes that the cursor of all buf's is zero,
54     *        and it can call buf_reset(source, tag).
55     */
56    int MPINU_rank_of_msg_avail_in_cache(int source, int tag) {
57      if (source != MPI_ANY_SOURCE) {      if (source != MPI_ANY_SOURCE) {
58        if (buf_avail(source, sizeof(struct msg_hdr)))        if (tag == MPI_ANY_TAG) {
59          return source;          if (buf_avail(source, sizeof(struct msg_hdr)))
60              return source;
61          }
62          else
63            while (buf_avail(source, sizeof(struct msg_hdr))) {
64              struct msg_hdr *hdr = buf_peek(source, sizeof(struct msg_hdr));
65              if ( ntohl( hdr->tag ) == tag ) {
66                buf_reset(source, tag); /* clean up for next caller */
67                return source;
68              } else { /* Else skip header and body */
69                buf_skip( source, sizeof(struct msg_hdr) + ntohl(hdr->size) );
70              }
71            }
72          buf_reset(source, tag); /* clean up for next caller */
73      }      }
74      else /* else source == MPI_ANY_SOURCE */      else /* else source == MPI_ANY_SOURCE */
75        for (source = 0; source < MPINU_num_slaves; source++)        for (source = 0; source < MPINU_num_slaves; source++) {
76          if (buf_avail(source, sizeof(struct msg_hdr)))          int rank = MPINU_rank_of_msg_avail_in_cache(source, tag);
77            return source;          if (rank != -1)
78              return rank;
79          }
80      /* if no source had buf_avail, then no msg avail */      /* if no source had buf_avail, then no msg avail */
81      return -1;      return -1;
82  }  }
83    
84  /* On receiving a msg_hdr, we first do buf_reset(source),  /* On receiving a msg_hdr, we first do buf_reset(source, tag),
85   * but _not_ on receiving a msg_body */   * but _not_ on receiving a msg_body */
86  ssize_t MPINU_recv_msg_hdr_with_cache(int s, int tag,  ssize_t MPINU_recv_msg_hdr_with_cache(int s, int tag,
87                                        void *buf, size_t len, int flags) {                                        void *buf, size_t len, int flags) {
88      int msg_found = 0;      int msg_found = 0;
89      int source = MPINU_rank_from_socket(s);      int source = MPINU_rank_from_socket(s);
90      buf_reset(source);      buf_reset(source, tag);
91      while ( !msg_found && buf_avail(source, len) ) {      while ( !msg_found && buf_avail(source, len) ) {
92        buf_peek(source, buf, len);        if ( ntohl( ((struct msg_hdr *)buf_peek(source, len))->tag ) == tag
       if ( ntohl( ((struct msg_hdr *)buf)->tag ) == tag  
93             || tag == MPI_ANY_TAG ) {             || tag == MPI_ANY_TAG ) {
94          buf_dequeue(source, buf, len, flags & MSG_PEEK);          buf_dequeue(source, buf, len, flags & MSG_PEEK);
95          msg_found = 1;          msg_found = 1;
# Line 140  static struct buf_queue { Line 161  static struct buf_queue {
161      void * buf;      void * buf;
162      int end;      int end;
163      int curs;      int curs;
164        int tag;
165  } * recv_buf;  } * recv_buf;
166    
167  static void buf_init () {  static void buf_init () {
# Line 155  static void buf_init () { Line 177  static void buf_init () {
177        recv_buf[i].curs = 0;        recv_buf[i].curs = 0;
178      }      }
179  }  }
180  static void buf_reset(int rank) {  /* Reset to search beginning of buffer, looking for tag */
181    static void buf_reset(int source, int tag) {
182        assert( source != MPI_ANY_SOURCE );
183      if ( recv_buf == NULL )      if ( recv_buf == NULL )
184        buf_init();        buf_init();
185      recv_buf[rank].curs = 0;      recv_buf[source].curs = 0;
186        recv_buf[source].tag = tag;
187  }  }
188  static int buf_avail (int rank, int len) {  static int buf_avail (int source, int len) {
189      if ( recv_buf == NULL ) /* if recv_buf never init'ed, then nothing avail */      if ( recv_buf == NULL ) /* if recv_buf never init'ed, then nothing avail */
190        return 0;        return 0;
191      return recv_buf[rank].curs + len <= recv_buf[rank].end;      return recv_buf[source].curs + len <= recv_buf[source].end;
192    }
193    /*
194     * If the tags are not equal, then check the entire buffer.
195     * If the tags are equal, buf_reset() was called for this tag.  Continue
196     *    searching starting at cursor.
197     */
198    static int buf_avail_tag(int source, int len, int tag) {
199        if ( recv_buf == NULL ) /* if recv_buf never init'ed, then nothing avail */
200          return 0;
201        assert( source != MPI_ANY_SOURCE );
202        assert( recv_buf[source].curs == 0 || recv_buf[source].tag != MPI_ANY_TAG);
203        /* I DON'T UNDERSTAND THIS FIRST CASE.  IF curs == 0, THEN BOTH
204           CASES DO THE SAME THING.  IF curs > 0, THEN THE TAGS DON'T MATCH,
205           AND SO THEN SHOULDN'T WE RETURN FALSE AND THEN FORCE A SEARCH FURTHER
206           INTO buf BY OUR CALLING FUNCTION?  - Gene */
207        if (buf_tag_compare(source, tag) == BUF_TAG_NOT_EQUAL)
208          return len <= recv_buf[source].end;
209        else
210          return recv_buf[source].curs + len <= recv_buf[source].end;
211    }
212    /*
213     * Compare the given tag with the tag of the cursor for the specified rank. If
214     * the cursor is at the beginning then declare
215     * that the tag does not match. Otherwise, if an actual tag comparison doesn't
216     * match, return that fact, else the tags do match.
217     */
218    static buf_tag_equality_t buf_tag_compare(int rank, int tag) {
219        assert( rank != MPI_ANY_SOURCE );
220        if ( recv_buf[rank].curs == 0 || recv_buf[rank].tag != tag )
221            return BUF_TAG_NOT_EQUAL;
222        else
223            return BUF_TAG_EQUAL;
224  }  }
225  static void buf_finalize () {  static void buf_finalize () {
226      free( recv_buf );      free( recv_buf );
# Line 198  static void buf_dequeue(int rank, void * Line 255  static void buf_dequeue(int rank, void *
255          free( recv_buf[rank].buf );          free( recv_buf[rank].buf );
256      }      }
257  }  }
258  static void buf_peek(int rank, void *buf, int len) {  static void *buf_peek(int rank, int len) {
259      assert( recv_buf[rank].curs + len <= recv_buf[rank].end );      assert( recv_buf[rank].curs + len <= recv_buf[rank].end );
260      memcpy( buf, recv_buf[rank].buf + recv_buf[rank].curs, len );      assert( recv_buf[rank].curs == 0 || recv_buf[rank].tag != MPI_ANY_TAG );
261        return (char *)recv_buf[rank].buf + recv_buf[rank].curs;
262  }  }
263  static void buf_skip(int rank, int len) {  static void buf_skip(int rank, int len) {
264      recv_buf[rank].curs += len;      recv_buf[rank].curs += len;

Legend:
Removed from v.1.1  
changed lines
  Added in v.1.2

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26