bsp.c

Go to the documentation of this file.

/*
    BSPonMPI. This is an implementation of the BSPlib standard on top of MPI
    Copyright (C) 2006  Wijnand J. Suijlen
                                                                                
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.
                                                                                
    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.
                                                                                
    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
                                                                                
    You may contact me via electronic mail:
      wjsuijle@users.sourceforge.net
    or snail mail:
      W.J. Suijlen
      Kraaiheidelaan 10
      2803 VP Gouda
      The Netherlands
*/

#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <mpi.h>
#include "bsp.h"
#include "bsp_memreg.h"
#include "bsp_mesgqueue.h"
#include "bsp_delivtable.h"
#include "bsp_reqtable.h"
#include "bsp_private.h"
#include "bsp_alloc.h"
#include "bsp_abort.h"
#include <config.h>

#define DELIVTAB_SIZE 1
#define REQTAB_SIZE   1
#define MEMREG_SIZE   1 


void
bsp_init (void (*spmd_part) (void), int argc, char *argv[])
{
  /* initialize MPI */
  MPI_Init(&argc, &argv);
  MPI_Comm_size( MPI_COMM_WORLD, &bsp.nprocs);
  MPI_Comm_rank( MPI_COMM_WORLD, &bsp.rank);

  if (bsp.rank == 0) 
    {
     /* do nothing */
    }
  else
    {
     /* else just run the spmd part */
      spmd_part();
      exit(0);
    }  
}

void
bsp_begin (int maxprocs)
{
  int flag, i, *ranks;
  MPI_Group group, newgroup;
  /* initialize if necessary */
  if (MPI_Initialized(&flag), !flag)
    {
      int argc = 0;
      char **argv = NULL;
      fprintf(stderr, "Warning! bsp_init() is not called. Initialization of MPI may fail\n");
      MPI_Init (&argc, &argv);
      MPI_Comm_size (MPI_COMM_WORLD, &bsp.nprocs);
      MPI_Comm_rank (MPI_COMM_WORLD, &bsp.rank);
    }
  /* broadcast maxprocs to all other processors */  
  MPI_Bcast(&maxprocs, 1, MPI_INT, 0, MPI_COMM_WORLD);
  
  /* allocate at most maxproc processors:
     Form a new group of processors. Some processors will not be member of
     this group and will therefore be deallocated */
  if (maxprocs > 0) 
    {
      bsp.nprocs = MIN(maxprocs, bsp.nprocs);
    } /* else request maximum number of processors*/
    
  MPI_Comm_group( MPI_COMM_WORLD, &group);
  ranks = bsp_malloc( bsp.nprocs, sizeof(int));
  for (i = 0; i < bsp.nprocs; i++)
    ranks[i] = i;

  MPI_Group_incl(group, bsp.nprocs, ranks, &newgroup);
  MPI_Comm_create(MPI_COMM_WORLD, newgroup, &bsp.communicator);

  bsp_free(ranks);

  if (bsp.rank >= bsp.nprocs)
    { /* terminate all unnecessary processes */
      MPI_Finalize();
      exit(0);
    }  

  /* initialize data structures */
  memoryRegister_initialize(&bsp.memory_register, bsp.nprocs, MEMREG_SIZE,
                            bsp.rank);
  messageQueue_initialize (&bsp.message_queue);
  deliveryTable_initialize(&bsp.delivery_table, bsp.nprocs, DELIVTAB_SIZE);
  requestTable_initialize(&bsp.request_table, bsp.nprocs, REQTAB_SIZE);
  deliveryTable_initialize(&bsp.delivery_received_table, bsp.nprocs, 
                           DELIVTAB_SIZE);
  requestTable_initialize(&bsp.request_received_table, bsp.nprocs,
                           REQTAB_SIZE);

  /* save starting time */
  bsp.begintime = MPI_Wtime ();
}


void
bsp_end ()
{
  /* clean up datastructures */
  memoryRegister_destruct (&bsp.memory_register);
  deliveryTable_destruct(&bsp.delivery_table);
  requestTable_destruct(&bsp.request_table);
  deliveryTable_destruct(&bsp.delivery_received_table);
  requestTable_destruct(&bsp.request_received_table);

  /* and finalize */
  MPI_Finalize ();
}

void
bsp_abort (const char *format, ...)
{
  va_list ap;
  va_start (ap, format);
  vfprintf (stderr, format, ap);
  va_end (ap);

  bsp_intern_abort (ERR_BSP_ABORT, "bsp_abort()", __FILE__, __LINE__);
}

int
bsp_nprocs ()
{
  int flag;
  MPI_Initialized(&flag);
  return flag?bsp.nprocs:-1;
}

int
bsp_pid ()
{
  return bsp.rank;
}

double
bsp_time ()
{
  return MPI_Wtime () - bsp.begintime;
}

void
bsp_sync ()
{
  unsigned int send_index[ 3 * bsp.nprocs] ;
  unsigned int recv_index[ 3 * bsp.nprocs] ;
  unsigned int maxreqrows = 0, maxdelrows = 0, p;
  unsigned int any_gets = 0; /* any_gets is a boolean value, whether there are
                               any gets to performed. If there are no gets,
                               then one MPI_Alltoall doesn't have to be
                               executed */
  /* reset message buffer */
  messageQueue_sync(&bsp.message_queue);
  requestTable_reset(&bsp.request_received_table);
  deliveryTable_reset(&bsp.delivery_received_table);
 
  /* communicate information */
  for (p = 0; p < bsp.nprocs; p++)
    any_gets |= bsp.request_table.used_slot_count[p];

  for (p = 0; p < bsp.nprocs; p++)
    {
      send_index[3*p    ] = bsp.request_table.used_slot_count[p];
      send_index[3*p + 1] = bsp.delivery_table.used_slot_count[p];
      send_index[3*p + 2] = any_gets;
    }  

  MPI_Alltoall( send_index, 3 , MPI_UNSIGNED, 
                recv_index, 3 , MPI_UNSIGNED, bsp.communicator);

  /* expand buffers if necessary */
  maxreqrows = max(recv_index, 3*bsp.nprocs, 3);
  for (p = 0; p < bsp.nprocs; p++)
    maxdelrows = MAX( recv_index[1 + 3*p] + 
                     bsp.request_table.info.req.data_sizes[p], maxdelrows);

  if ( bsp.request_received_table.rows < maxreqrows )
    {
      maxreqrows = MAX(bsp.request_received_table.rows, maxreqrows);
      requestTable_expand(&bsp.request_received_table, maxreqrows);
    }  
  
  if (bsp.delivery_received_table.rows < maxdelrows )
    {
      maxdelrows = MAX(bsp.delivery_received_table.rows, maxdelrows);
      deliveryTable_expand(&bsp.delivery_received_table, maxdelrows );
    }  

  /* copy necessary indices to received_tables */
  for (p = 0; p < bsp.nprocs; p++) 
    {
      bsp.request_received_table.used_slot_count[p] = recv_index[3*p];
      bsp.delivery_received_table.used_slot_count[p] =
        recv_index[1 + 3*p] + bsp.request_table.info.req.data_sizes[p] ;
    }   
  
  /* Now we may conclude something about the communcation pattern */
  any_gets = 0;
  for (p = 0; p < bsp.nprocs; p++)   
    any_gets |= recv_index[3*p + 2];

  /* communicate & execute */
  if (any_gets) 
    {
      expandableTable_comm(&bsp.request_table, &bsp.request_received_table,
                    bsp.communicator);
      requestTable_execute(&bsp.request_received_table, &bsp.delivery_table);
    }

  expandableTable_comm(&bsp.delivery_table, &bsp.delivery_received_table,
                     bsp.communicator);
  deliveryTable_execute(&bsp.delivery_received_table, 
                        &bsp.memory_register, &bsp.message_queue, bsp.rank);
  /* clear the buffers */                       
  requestTable_reset(&bsp.request_table);
  deliveryTable_reset(&bsp.delivery_table);
 
  /* pack the memoryRegister */
  memoryRegister_pack(&bsp.memory_register);
}
void
bsp_push_reg (const void *ident, int size)
{
  int i;
  DelivElement element;
  element.size = 0;
  element.info.push.address = ident;
  for (i=0 ; i < bsp.nprocs; i++)
    deliveryTable_push(&bsp.delivery_table, i, element, pushreg);
}

void
bsp_pop_reg (const void *ident)
{
  DelivElement element;
  element.size = 0;
  element.info.pop.address = ident;
  deliveryTable_push(&bsp.delivery_table, bsp.rank, element, popreg);
}  

void
bsp_put (int pid, const void *src, void *dst, int offset, int nbytes)
{
  /* place put command in buffer */
  char * restrict pointer;
  DelivElement element;
  element.size = nbytes;
  element.info.put.dst = 
    memoryRegister_memoized_find(&bsp.memory_register, pid, dst) + offset;
  pointer = deliveryTable_push(&bsp.delivery_table, pid, element, put);
  memcpy(pointer, src, nbytes);
}


void
bsp_get (int pid, const void *src, int offset, void *dst, int nbytes)
{
  ReqElement elem;
  elem.size = nbytes;
  elem.src = 
     memoryRegister_memoized_find(&bsp.memory_register, pid, src);
  elem.dst = dst;
  elem.offset = offset;
  
  /* place get command in buffer */
  requestTable_push(&bsp.request_table, pid, elem);
}

void
bsp_send (int pid, const void *tag, const void *payload, int payload_nbytes)
{
  DelivElement element;
  char * restrict pointer;
  element.size = payload_nbytes + bsp.message_queue.send_tag_size;
  element.info.send.payload_size = payload_nbytes;
  pointer = deliveryTable_push(&bsp.delivery_table, pid, element, send);
  memcpy( pointer, tag, bsp.message_queue.send_tag_size);
  memcpy( pointer + bsp.message_queue.send_tag_size, payload, payload_nbytes);
}

void
bsp_qsize (int * restrict nmessages, int * restrict accum_nbytes)
{
  *nmessages = bsp.message_queue.n_mesg;
  *accum_nbytes = bsp.message_queue.accum_size;
}

void
bsp_get_tag (int * restrict status , void * restrict tag)
{
  if (bsp.message_queue.n_mesg == 0)
    *status = -1;
  else
    {
      ALIGNED_TYPE * restrict current_tag = 
        bsp.message_queue.head + 
         no_slots( sizeof(DelivElement), sizeof(ALIGNED_TYPE));
      DelivElement * restrict message = (DelivElement *) bsp.message_queue.head;        
      *status = message->size;
      memcpy(tag, current_tag, bsp.message_queue.recv_tag_size ); 
    }
}

void
bsp_move (void *payload, int reception_nbytes)
{
  DelivElement * restrict message = (DelivElement *) bsp.message_queue.head;    
  int copy_bytes = MIN(reception_nbytes, message->size);
  char * restrict current_payload =
    (char *) bsp.message_queue.head + 
      sizeof(ALIGNED_TYPE) * 
      no_slots( sizeof(DelivElement), sizeof(ALIGNED_TYPE)) +
    bsp.message_queue.recv_tag_size;
  memcpy(payload, current_payload, copy_bytes);
  
  bsp.message_queue.head += message->next;
  bsp.message_queue.n_mesg --;
  bsp.message_queue.accum_size -= message->size;
}

void
bsp_set_tagsize (int *tag_nbytes)
{
  DelivElement element;
  element.info.settag.tag_size = *tag_nbytes;
  element.size = 0;
 
  deliveryTable_push(&bsp.delivery_table, bsp.rank, element, settag);
  *tag_nbytes = bsp.message_queue.send_tag_size;
}


void
bsp_hpput (int pid, const void *src, void *dst, int offset, int nbytes)
{
  bsp_put (pid, src, dst, offset, nbytes);
}
void
bsp_hpget (int pid, const void *src, int offset, void *dst, int nbytes)
{
  bsp_get (pid, src, offset, dst, nbytes);
}

int
bsp_hpmove (void **tag_ptr, void **payload_ptr)
{
  if (bsp.message_queue.n_mesg == 0)
    return -1;
  else
    {
       ALIGNED_TYPE * restrict current_tag =
         (ALIGNED_TYPE *) bsp.message_queue.head + 
           no_slots(sizeof(DelivElement), sizeof(ALIGNED_TYPE));
       char * restrict current_payload =
         (char *) current_tag + bsp.message_queue.recv_tag_size;
       DelivElement * restrict message =
         (DelivElement *) bsp.message_queue.head;
       int size = message->info.send.payload_size;       
       *tag_ptr     = current_tag;
       *payload_ptr = current_payload;
       bsp.message_queue.head += message->next;
       bsp.message_queue.n_mesg--;
       bsp.message_queue.accum_size -= size;
       return size;
    }
}

---