mpid_recvq.c

fortran并行计算包

                   )
                {
                    if (prev_rreq != NULL)
                    {
                        prev_rreq->dcmf.next = rreq->dcmf.next;
                    }
                    else
                    {
                        recvq.unexpected_head = rreq->dcmf.next;
                    }
                    if (rreq->dcmf.next == NULL)
                    {
                        recvq.unexpected_tail = prev_rreq;
                    }
                    found = TRUE;
                    goto lock_exit;
                }

                prev_rreq = rreq;
                rreq = rreq->dcmf.next;
            }
        }
        else
        {
            MPIDI_Message_match match;
            MPIDI_Message_match mask;

            match.context_id = context_id;
            mask.context_id = ~0;
            if (tag == MPI_ANY_TAG)
            {
                match.tag = 0;
                mask.tag = 0;
            }
            else
            {
                match.tag = tag;
                mask.tag = ~0;
            }
            if (source == MPI_ANY_SOURCE)
            {
                match.rank = 0;
                mask.rank = 0;
            }
            else
            {
                match.rank = source;
                mask.rank = ~0;
            }

            prev_rreq = NULL;
            rreq = recvq.unexpected_head;
            while (rreq != NULL)
            {
#ifdef USE_STATISTICS
                ++search_length;
#endif
                if ( (  rreq->dcmf.msginfo.msginfo.MPIctxt              == match.context_id) &&
                     ( (rreq->dcmf.msginfo.msginfo.MPIrank & mask.rank) == match.rank      ) &&
                     ( (rreq->dcmf.msginfo.msginfo.MPItag  & mask.tag ) == match.tag       )
                   )
                {
                    if (prev_rreq != NULL)
                    {
                        prev_rreq->dcmf.next = rreq->dcmf.next;
                    }
                    else
                    {
                        recvq.unexpected_head = rreq->dcmf.next;
                    }
                    if (rreq->dcmf.next == NULL)
                    {
                        recvq.unexpected_tail = prev_rreq;
                    }
                    found = TRUE;
                    goto lock_exit;
                }

                prev_rreq = rreq;
                rreq = rreq->dcmf.next;
            }
        }

        /* A matching request was not found in the unexpected queue,
           so we need to allocate a new request and add it to the
           posted queue */
        rreq = MPID_Request_create();
        if (rreq != NULL)
        {
            MPIU_Object_set_ref(rreq, 2);
            rreq->kind = MPID_REQUEST_RECV;
            rreq->dcmf.msginfo.msginfo.MPItag = tag;
            rreq->dcmf.msginfo.msginfo.MPIrank = source;
            rreq->dcmf.msginfo.msginfo.MPIctxt = context_id;
            rreq->dcmf.next = NULL;

            if (recvq.posted_tail != NULL)
            {
                recvq.posted_tail->dcmf.next = rreq;
            }
            else
            {
                recvq.posted_head = rreq;
            }
            recvq.posted_tail = rreq;
        }

        found = FALSE;
    }
  lock_exit:
    MPIDI_Recvq_unlock();

#ifdef USE_STATISTICS
    MPIDI_Statistics_time(MPIDI_Statistics.recvq.unexpected_search, search_length);
#endif

    *foundp = found;
    return rreq;
}


/**
 * \brief Find a request in the posted queue and dequeue it
 * \param [in]  req        Find by address of request object on sender
 * \return      The matching posted request or NULL
 */
int MPIDI_Recvq_FDPR(MPID_Request * req)
{
    MPID_Request * cur_rreq  = NULL;
    MPID_Request * prev_rreq = NULL;
    int found = FALSE;
#ifdef USE_STATISTICS
    unsigned search_length = 0;
#endif

    MPIDI_Recvq_lock();
    {
        cur_rreq = recvq.posted_head;
        while (cur_rreq != NULL)
        {
#ifdef USE_STATISTICS
            ++search_length;
#endif
            if (cur_rreq == req)
            {
                if (prev_rreq != NULL)
                {
                    prev_rreq->dcmf.next = cur_rreq->dcmf.next;
                }
                else
                {
                    recvq.posted_head = cur_rreq->dcmf.next;
                }
                if (cur_rreq->dcmf.next == NULL)
                {
                    recvq.posted_tail = prev_rreq;
                }

                found = TRUE;
                break;
            }

            prev_rreq = cur_rreq;
            cur_rreq = cur_rreq->dcmf.next;
        }
    }
    MPIDI_Recvq_unlock();

#ifdef USE_STATISTICS
    MPIDI_Statistics_time(MPIDI_Statistics.recvq.posted_search, search_length);
#endif

    return found;
}


/**
 * \brief Find a request in the posted queue and dequeue it, or allocate a new request and enqueue it in the unexpected queue
 * \param [in]  source     Find by Sender
 * \param [in]  tag        Find by Tag
 * \param [in]  context_id Find by Context ID (communicator)
 * \param [out] foundp    TRUE iff the request was found
 * \return      The matching posted request or the new UE request
 */
MPID_Request * MPIDI_Recvq_FDP_or_AEU(int source, int tag, int context_id, int * foundp)
{
    MPID_Request * rreq;
    MPID_Request * prev_rreq = NULL;
    int found = FALSE;
#ifdef USE_STATISTICS
    unsigned search_length = 0;
#endif


    MPIDI_Recvq_lock();
    {
        rreq = recvq.posted_head;
        while (rreq != NULL)
        {
#ifdef USE_STATISTICS
            ++search_length;
#endif
            if ( (rreq->dcmf.msginfo.msginfo.MPIctxt == context_id) &&
                 (rreq->dcmf.msginfo.msginfo.MPIrank == source || rreq->dcmf.msginfo.msginfo.MPIrank == MPI_ANY_SOURCE) &&
                 (rreq->dcmf.msginfo.msginfo.MPItag  == tag    || rreq->dcmf.msginfo.msginfo.MPItag  == MPI_ANY_TAG)
               )
            {
                if (prev_rreq != NULL)
                {
                    prev_rreq->dcmf.next = rreq->dcmf.next;
                }
                else
                {
                    recvq.posted_head = rreq->dcmf.next;
                }
                if (rreq->dcmf.next == NULL)
                {
                    recvq.posted_tail = prev_rreq;
                }
                found = TRUE;
                goto lock_exit;
            }

            prev_rreq = rreq;
            rreq = rreq->dcmf.next;
        }

        /* A matching request was not found in the posted queue, so we
           need to allocate a new request and add it to the unexpected
           queue */
        rreq = MPID_Request_create();
        if (rreq != NULL)
        {
            MPIU_Object_set_ref(rreq, 2);
            rreq->kind = MPID_REQUEST_RECV;
            MPID_Request_setMatch(rreq, tag, source, context_id);
            rreq->dcmf.next = NULL;

            if (recvq.unexpected_tail != NULL)
            {
                recvq.unexpected_tail->dcmf.next = rreq;
            }
            else
            {
                recvq.unexpected_head = rreq;
            }
            recvq.unexpected_tail = rreq;
        }

        found = FALSE;
    }
  lock_exit:
    MPIDI_Recvq_unlock();

#ifdef USE_STATISTICS
    MPIDI_Statistics_time(MPIDI_Statistics.recvq.posted_search, search_length);
#endif

    *foundp = found;
    return rreq;
}


/**
 * \brief Dump the queues
 */
void MPIDI_Recvq_DumpQueues (int verbose)
{
  if(verbose == 0)
    return;

  MPID_Request * rreq = recvq.posted_head;
  MPID_Request * prev_rreq = NULL;
  unsigned i=0, numposted=0, numue=0;
  unsigned postedbytes=0, uebytes=0;

  fprintf(stderr,"Posted Queue:\n");
  fprintf(stderr,"-------------\n");
  while (rreq != NULL)
    {
      if(verbose >= 2)
        fprintf (stderr, "P %d: MPItag=%d MPIrank=%d ctxt=%d cc=%d count=%d\n",
                 i++,
                 rreq->dcmf.msginfo.msginfo.MPItag,
                 rreq->dcmf.msginfo.msginfo.MPIrank,
                 rreq->dcmf.msginfo.msginfo.MPIctxt,
                 rreq->cc,
                 rreq->dcmf.userbufcount
                 );
      numposted++;
      postedbytes+=rreq->dcmf.userbufcount;
      prev_rreq = rreq;
      rreq = rreq->dcmf.next;
    }
  fprintf(stderr, "Posted Requests %d, Total Mem: %d bytes\n",
          numposted, postedbytes);


  i=0;
  rreq = recvq.unexpected_head;
  fprintf(stderr, "Unexpected Queue:\n");
  fprintf(stderr, "-----------------\n");
  while (rreq != NULL)
    {
      if(verbose >= 2)
        fprintf (stderr, "UE %d: MPItag=%d MPIrank=%d ctxt=%d cc=%d uebuf=%p uebuflen=%u\n",
                 i++,
                 rreq->dcmf.msginfo.msginfo.MPItag,
                 rreq->dcmf.msginfo.msginfo.MPIrank,
                 rreq->dcmf.msginfo.msginfo.MPIctxt,
                 *rreq->cc_ptr,
                 rreq->dcmf.uebuf,
                 rreq->dcmf.uebuflen);
      numue++;
      uebytes+=rreq->dcmf.uebuflen;
      prev_rreq = rreq;
      rreq = rreq->dcmf.next;
    }
  fprintf(stderr, "Unexpected Requests %d, Total Mem: %d bytes\n",
          numue, uebytes);

}