xmp_async.c File Reference

#include "xmp_internal.h"
#include <string.h>

Include dependency graph for xmp_async.c:

Functions
_Bool	xmp_is_async ()
_XMP_async_comm_t *	_XMP_get_async (int async_id)
void	_XMP_initialize_async_comm_tab ()
void	_XMP_wait_async__ (int async_id)
int	xmp_test_async_ (int *async_id)
_XMP_async_comm_t *	_XMP_get_current_async ()
void	xmpc_init_async (int async_id)
void	_XMP_nodes_dealloc_after_wait_async (_XMP_nodes_t *n)
void	xmpc_start_async ()
void	xmpc_end_async (int async_id)

Variables
void(*	_XMP_unpack_comm_set )(void *recvbuf, int recvbuf_size, _XMP_array_t *a, _XMP_comm_set_t *comm_set[][_XMP_N_MAX_DIM])

Function Documentation

_XMP_get_async()

_XMP_async_comm_t* _XMP_get_async

(

int

async_id

)

{
  int hash_id = async_id % _XMP_ASYNC_COMM_SIZE;
  _XMP_async_comm_t *async = &_XMP_async_comm_tab[hash_id];
  
  if(async->async_id == async_id)  return async;
 
  while(async->next){
    async = async->next;
    if(async->async_id == async_id)
      return async;
  }
 
  return NULL;
}

Here is the caller graph for this function:

_XMP_get_current_async()

_XMP_async_comm_t* _XMP_get_current_async

(

)

{
  return _tmp_async;
}

Here is the caller graph for this function:

_XMP_initialize_async_comm_tab()

void _XMP_initialize_async_comm_tab

(

)

{
  for(int i=0;i<_XMP_ASYNC_COMM_SIZE;i++){
    _XMP_async_comm_tab[i].nreqs   = 0;
    _XMP_async_comm_tab[i].nnodes  = 0;
    _XMP_async_comm_tab[i].is_used = false;
    _XMP_async_comm_tab[i].type    = _XMP_COMM_NONE;
    _XMP_async_comm_tab[i].node    = NULL;
    _XMP_async_comm_tab[i].reqs    = NULL;
    _XMP_async_comm_tab[i].gmove   = NULL;
    _XMP_async_comm_tab[i].a       = NULL;
    _XMP_async_comm_tab[i].next    = NULL;
  }
}

_XMP_nodes_dealloc_after_wait_async()

void _XMP_nodes_dealloc_after_wait_async

(

_XMP_nodes_t *

n

)

{
  if(_tmp_async->nnodes >= _XMP_MAX_ASYNC_NODES) // Fix me
    _XMP_fatal("Too many nodes");
 
  if(_tmp_async->nnodes == 0)
    _tmp_async->node = _XMP_alloc(sizeof(_XMP_nodes_t*) * _XMP_MAX_ASYNC_NODES);
  
  _tmp_async->node[_tmp_async->nnodes] = n;
  _tmp_async->nnodes++;
}

Here is the caller graph for this function:

_XMP_wait_async__()

void _XMP_wait_async__

(

int

async_id

)

{
  _XMP_async_comm_t *async;
 
  if(!(async = _XMP_get_async(async_id))) return;
 
  int nreqs         = async->nreqs;
  MPI_Request *reqs = async->reqs;
 
#if defined(_KCOMPUTER) && defined(K_RDMA_REFLECT)
  // For RDMA reflects, async->nreqs > 0 and async->reqs == NULL.
  if(nreqs && !reqs){
    _XMP_wait_async_rdma(async);
    return;
  }
#endif
 
  _XMP_async_gmove_t *gmove = async->gmove;
 
  if (async->type == _XMP_COMM_REDUCE_SHADOW){
    MPI_Waitall(nreqs, reqs, MPI_STATUSES_IGNORE);
    _XMP_reduce_shadow_sum(async->a);
  }
  else if (!gmove || gmove->mode == _XMP_N_GMOVE_NORMAL){
    _XMP_TSTART(t0);
    MPI_Waitall(nreqs, reqs, MPI_STATUSES_IGNORE);
    _XMP_TEND(xmptiming_.t_wait, t0);
  }
 
  //
  // for asynchronous gmove
  //
 
  if (gmove){
    if (gmove->mode == _XMP_N_GMOVE_NORMAL) _XMP_finalize_async_gmove(gmove);
#ifdef _XMP_MPI3_ONESIDED
    else {
      int status;
      // NOTE: the send_buf field is used for an improper purpose.
      _XMP_sync_images_COMM((MPI_Comm *)gmove->sendbuf, &status);
    }
#endif
  }
 
}

Here is the call graph for this function:

xmp_is_async()

_Bool xmp_is_async

(

)

{
  return _xmp_is_async;
}

Here is the caller graph for this function:

xmp_test_async_()

int xmp_test_async_

(

int *

async_id

)

{
  _XMP_async_comm_t *async;
 
  if (!(async = _XMP_get_async(*async_id))) return 1;
 
  int nreqs = async->nreqs;
  MPI_Request *reqs = async->reqs;
 
#if defined(_KCOMPUTER) && defined(K_RDMA_REFLECT)
  // For RDMA reflects, async->nreqs > 0 and async->reqs == NULL.
  _XMP_fatal("xmp_test_async not supported for RDMA.");
#endif
 
  int flag;
  MPI_Testall(nreqs, reqs, &flag, MPI_STATUSES_IGNORE);
 
  if(flag){
    _XMP_async_gmove_t *gmove = async->gmove;
    if (gmove) _XMP_finalize_async_gmove(gmove);
    else if(async->type == _XMP_COMM_REDUCE_SHADOW)
      _XMP_reduce_shadow_sum(async->a);
 
    xmpc_end_async(*async_id);
    return 1;
  }
  else{
    return 0;
  }
}

Here is the call graph for this function:

xmpc_end_async()

void xmpc_end_async

(

int

async_id

)

{
  int hash_id = async_id % _XMP_ASYNC_COMM_SIZE;
  _XMP_async_comm_t *async = &_XMP_async_comm_tab[hash_id];
  
  if(!async->is_used) return;
 
  // The case no comm. registered for async_id 0 and xmpc_end_async called for
  // async_id == 0, may occur and is inconsistent.
  // But, actually, the code below works without problems in such a case, even if
  // no hash_id == 0.
  if(async->async_id == async_id){
    if(async->next == NULL){
      initialize_async(async);
    }
    else{
      _XMP_async_comm_t *next = async->next;
      initialize_async(async);
      async->async_id = next->async_id;
      async->nreqs    = next->nreqs;
      async->nnodes   = next->nnodes;
      async->is_used  = next->is_used;
      async->type     = next->type;
      async->node     = next->node;
      async->gmove    = next->gmove;
      async->reqs     = next->reqs;
      async->a        = next->a;
      async->next     = next->next;
      _XMP_free(next);
    }
    return;
  }
  else{
    _XMP_async_comm_t *prev = async;
    while((async = prev->next)){
      if(async->async_id == async_id){
        prev->next = async->next;
        initialize_async(async);
        _XMP_free(async);
        return;
      }
      prev = async;
    }
  }
}

xmpc_init_async()

void xmpc_init_async

(

int

async_id

)

{
  _xmp_is_async = true;
 
  int hash_id = async_id % _XMP_ASYNC_COMM_SIZE;
  _XMP_async_comm_t *async = &_XMP_async_comm_tab[hash_id];
 
  if(!async->is_used){
    async->async_id = async_id;
    async->reqs     = _XMP_alloc(sizeof(MPI_Request) * _XMP_MAX_ASYNC_REQS);
    async->is_used  = true;
    _tmp_async      = async;
  }
  else{
    if(async->async_id == async_id){
      _tmp_async = async;
    }
    else{
      while(async->next){
        async = async->next;
        if(async->async_id == async_id){
          _tmp_async = async;
          return;
        }
      }
 
      // When above while-statement cannot find a async descriptor,
      // the following lines create a new async descriptor
      async->next     = _XMP_alloc(sizeof(_XMP_async_comm_t));
      async           = async->next;
      async->async_id = async_id;
      async->nreqs    = 0;
      async->nnodes   = 0;
      async->is_used  = true;
      async->type     = _XMP_COMM_NONE;
      async->node     = NULL;
      async->reqs     = _XMP_alloc(sizeof(MPI_Request) * _XMP_MAX_ASYNC_REQS);
      async->gmove    = NULL;
      async->a        = NULL;
      async->next     = NULL;
      _tmp_async      = async;
    }
  }
}

xmpc_start_async()

void xmpc_start_async

(

)

{
  _xmp_is_async = false;
  _tmp_async    = NULL;
}

Variable Documentation

_XMP_unpack_comm_set

void(* _XMP_unpack_comm_set) (void *recvbuf, int recvbuf_size, _XMP_array_t *a, _XMP_comm_set_t *comm_set[][_XMP_N_MAX_DIM])