src/common/parallel_com.cc

/****************************
**
**      parallel_com.cc
**      Parallel Communication
**              by Kyle Ellrott
**
**
*****************************/

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>

#include "parallel_com.h"

#ifdef __comm_selected


#ifdef _USE_MPI
int rank, size;
#endif


int parallel_init(int *argc, char ***argv) {
  int status;
#ifdef _USE_PVM
  if (parent_tid == PvmNoParent) {
    for (n = 0; n < nworkers; n++) {
      rcode = pvm_spawn("calcpi", NULL, PvmTaskDefault, "", nworkers, tids[n]);
    }
    
  }
#endif
#ifdef _USE_MPI
  status = MPI_Init( argc, argv );
  if (status == MPI_SUCCESS) {
    //    fprintf(stderr, "Running MPI\n");
    MPI_Comm_size ( MPI_COMM_WORLD, &size ); 
    MPI_Comm_rank ( MPI_COMM_WORLD, &rank ); 
    return 0;
  }
  return 1;

#endif

  return 0;
}


int parallel_close(void) {
#ifdef _USE_PVM
  
#endif
#ifdef _USE_MPI
  MPI_Finalize();
#endif
}

int parallel_master_task(void) {
#ifdef _USE_MPI
  if (rank == 0) 
    return 1;
  else 
    return 0;
#endif
  return 0;
}

int parallel_slave_task(void) {
#ifdef _USE_MPI
  if (rank != 0) 
    return 1;
  else 
    return 0;
#endif
  return 0;
  
}


int parallel_group_size(void) {
#ifdef _USE_MPI
  return size;
#endif
}


int parallel_send_text(char *str, int dest, int tag) {
#ifdef _USE_MPI
  int size = strlen(str)+1;
  MPI_Send( &size, 1, MPI_INT, dest, tag, MPI_COMM_WORLD);
  return MPI_Send(str, size, MPI_CHAR, dest, tag, MPI_COMM_WORLD);
#endif
}



int parallel_recv_text(char **str, int src, int tag) {
#ifdef _USE_MPI
  int size;
  MPI_Status status;
  MPI_Recv( &size, 1, MPI_INT, src, tag, MPI_COMM_WORLD, &status);
  *str = (char *)malloc( size );
   memset(*str, 0, size-1);
  return MPI_Recv( *str, size, MPI_CHAR, src, tag, MPI_COMM_WORLD, &status);
#endif
}

int parallel_send_data(void *data, int data_size, int dest, int tag) {
#ifdef _USE_MPI
  MPI_Send( &data_size, 1, MPI_INT, dest, tag, MPI_COMM_WORLD);
  return MPI_Send(data, data_size, MPI_CHAR, dest, tag, MPI_COMM_WORLD);
#endif
}


int parallel_recv_data(void **data, int *data_size, int src, int tag) {
#ifdef _USE_MPI
  MPI_Status status;
  MPI_Recv( data_size, 1, MPI_INT, src, tag, MPI_COMM_WORLD, &status);
  *data = malloc( *data_size );
  memset(*data, 0, (*data_size)-1);
  return MPI_Recv( *data, *data_size, MPI_CHAR, src, tag, MPI_COMM_WORLD, &status);
#endif
}



int parallel_send_int(int num, int dest, int tag) {
#ifdef _USE_MPI
  return MPI_Send( &num, 1, MPI_INT, dest, tag, MPI_COMM_WORLD);
#endif
}



int parallel_recv_int(int *num, int src, int tag) {
#ifdef _USE_MPI
  MPI_Status status;
  return MPI_Recv( num, 1, MPI_INT, src, tag, MPI_COMM_WORLD, &status);
#endif
}



int parallel_probe( int *src_val, int *tag_val) {
#ifdef _USE_MPI
  MPI_Status status;
  int flag;
  int src_num, tag_num;
  
  if (src_val == NULL)  
    src_num = MPI_ANY_SOURCE;
  else 
    src_num = *src_val;
  if (tag_val == NULL)
    tag_num = MPI_ANY_TAG;
  else 
    tag_num = *tag_val;

  MPI_Iprobe( src_num, tag_num, MPI_COMM_WORLD, &flag, &status );
  
  if ( flag ) {
    if ( src_val != NULL )
      *src_val = status.MPI_SOURCE;
    if ( tag_val != NULL ) 
      *tag_val = status.MPI_TAG;
    return 1;
  }
  return 0;
#endif
}



#define kRequestTag             1000
#define kWorkTag                1050
#define kResultTag              1100
#define kRetryTag               1125
#define kEndTag                 1150


#define tsWaiting               0
#define tsWorking               1
#define tsDone                  2
#define tsRetry                 3


void parallel_master_work(int work_count, 
                      request_function req_func, 
                      result_function res_func, 
                      void *data) {
  int *work_status, i, tmp_src, tmp_tag, working, cur_work;
  void *buffer_work, *buffer_result;
  int buffer_work_size, buffer_result_size;

  work_status = (int *) malloc(sizeof(int) * work_count );
  for ( i = 0; i < work_count; i++ ) {
    work_status[i] = tsWaiting;
  }

  MPI_Barrier(MPI_COMM_WORLD);
  working = work_count;

  do {
    tmp_src = PARALLEL_ANY_SOURCE;
    tmp_tag = kRequestTag;
    if ( parallel_probe(&tmp_src, &tmp_tag)) {
      parallel_recv_int(&i, tmp_src, tmp_tag);
      tmp_tag = kWorkTag;
      for (i = 0; i < work_count; i++) {
        if ( work_status[i] == tsWaiting) {
          cur_work = i;
          buffer_work_size = req_func( cur_work, data, &buffer_work );
          if ( buffer_work_size > 0) {
            i = work_count;
          } else {
            work_status[i] = tsDone;
            working--;
          }       
        }
      }
      if (  buffer_work_size > 0) {
        parallel_send_int( cur_work, tmp_src, tmp_tag);
        parallel_send_data( buffer_work, buffer_work_size, tmp_src, tmp_tag);
        work_status[ cur_work ] = tsWorking;
        free(buffer_work);
      }
    }
    tmp_src = PARALLEL_ANY_SOURCE;
    tmp_tag = kResultTag;
    while ( parallel_probe(&tmp_src, &tmp_tag)) {
      parallel_recv_int(&i, tmp_src, tmp_tag);
      parallel_recv_data(&buffer_result, &buffer_result_size, tmp_src, tmp_tag);
      work_status[ i ] = tsDone;
      res_func(  i, data, buffer_result, buffer_result_size);
      working--;
      free(buffer_result);
      tmp_src = PARALLEL_ANY_SOURCE;
      tmp_tag = kResultTag;
    }
    tmp_src = PARALLEL_ANY_SOURCE;
    tmp_tag = kRetryTag;
    while ( parallel_probe(&tmp_src, &tmp_tag)) {
      parallel_recv_int(&i, tmp_src, tmp_tag);
      work_status[ i ] = tsWaiting;
    }
    usleep(1000);
  } while (working);
  for (i = 1; i <  parallel_group_size(); i++) 
    parallel_send_int(0, i, kEndTag );
  MPI_Barrier(MPI_COMM_WORLD);
}


void parallel_slave_work(work_function work_func, void *data) {
  int tmp_src, tmp_tag, buffer_int;
  void *buffer_request, *buffer_result;
  char working = 1;
  int buffer_request_size, buffer_result_size;

  MPI_Barrier(MPI_COMM_WORLD);

  parallel_send_int(1, 0, kRequestTag );
  do {
    tmp_src = PARALLEL_ANY_SOURCE;
    tmp_tag = kWorkTag;
    if ( parallel_probe( &tmp_src, &tmp_tag ) ) {
      parallel_recv_int( &buffer_int, tmp_src, tmp_tag);
      parallel_recv_data( &buffer_request, &buffer_request_size, tmp_src, tmp_tag);
      
      buffer_result_size = work_func(buffer_int, data, buffer_request, buffer_request_size, &buffer_result);

      if (buffer_result_size == -1) {
        parallel_send_int( buffer_int, tmp_src, kRetryTag );
        parallel_send_int( 1, 0, kRequestTag );
      } else {
        parallel_send_int( buffer_int, tmp_src, kResultTag );
        parallel_send_data( buffer_result, buffer_result_size, tmp_src, kResultTag );
        parallel_send_int( 1, 0, kRequestTag );
      }
    }
    tmp_src = PARALLEL_ANY_SOURCE;
    tmp_tag = kEndTag;
    if ( parallel_probe( &tmp_src, &tmp_tag ) ) {
      parallel_recv_int( &buffer_int, tmp_src, tmp_tag);
      working = 0;
    }
    usleep(1000);
  } while (working);
  MPI_Barrier(MPI_COMM_WORLD);
}




#endif

---