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 1000
#define REQTAB_SIZE   1000
#define MEMREG_SIZE   100       
#define MESGQ_SIZE    1000


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, MESGQ_SIZE);
  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);
  messageQueue_destruct (&bsp.message_queue);
  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 ()
{
  /* reset message buffer */
  messageQueue_reset(&bsp.message_queue);

  /* communicate & execute */
  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);
  /* clear the buffers */                       
  expandableTable_reset(&bsp.request_table);
  expandableTable_reset(&bsp.request_received_table);
  deliveryTable_reset(&bsp.delivery_table);
  deliveryTable_reset(&bsp.delivery_received_table);
 
  /* pack the memoryRegister */
  memoryRegister_pack(&bsp.memory_register);
}
void
bsp_push_reg (const void *ident, int size)
{
  int i;
  VarSizeElement elem;
  elem.size = 0;
  elem.type = pushreg;
  elem.info.push.address = ident;
  elem.data = NULL;
  for (i=0 ; i < bsp.nprocs; i++)
    deliveryTable_push(&bsp.delivery_table, i, elem);
}

void
bsp_pop_reg (const void *ident)
{
  VarSizeElement elem;
  elem.size = 0;
  elem.type = popreg;
  elem.info.pop.address = ident;
  elem.data = NULL;
  deliveryTable_push(&bsp.delivery_table, bsp.rank, elem);
}  

void
bsp_put (int pid, const void *src, void *dst, int offset, int nbytes)
{
  /* place put command in buffer */
  deliveryTable_pushPut(&bsp.delivery_table, pid, 
     memoryRegister_memoized_find(&bsp.memory_register, pid, dst) + offset,
     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)
{
  deliveryTable_pushSend(&bsp.delivery_table, pid, tag, 
       messageQueue_getTagSize(&bsp.message_queue), payload, payload_nbytes);
}

void
bsp_qsize (int * restrict nmessages, int * restrict accum_nbytes)
{
  *nmessages = messageQueue_getNumber (&bsp.message_queue);
  *accum_nbytes = messageQueue_getAccumSize (&bsp.message_queue);
}

void
bsp_get_tag (int * restrict status , void * restrict tag)
{
  int nmesg = messageQueue_getNumber (&bsp.message_queue);
  if (nmesg == 0)
    *status = -1;
  else
    *status = messageQueue_getTag (&bsp.message_queue, tag);
}

void
bsp_move (void *payload, int reception_nbytes)
{
  messageQueue_dequeue (&bsp.message_queue, payload, reception_nbytes);
}

void
bsp_set_tagsize (int *tag_nbytes)
{
  VarSizeElement elem;
  elem.info.settag.tag_size = *tag_nbytes;
  elem.size = 0;
  elem.type = settag;
  elem.data = NULL;
 
  deliveryTable_push(&bsp.delivery_table, bsp.rank, elem);
  *tag_nbytes = messageQueue_getTagSize (&bsp.message_queue);
}


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 (const void **tag_ptr, const void **payload_ptr)
{
  if (messageQueue_getNumber (&bsp.message_queue) == 0)
    return -1;
  else
    return messageQueue_hpdequeue(&bsp.message_queue, tag_ptr, payload_ptr);
}

---