mpi_pvm.c

Parallel programming／Lou Baker, Bradley J.Smith ．—New York：McGraw-Hill Book Co.

/* PSPH - Parallel SPH program
 * Bradley Smith, and Lou Baker, Dagonet Software
 */
static char sccs_id[] = "@(#) /home2/bsmith/src/psph/SCCS/s.mpi_pvm.c 1.17 94/03/30";

/* MPI Implementation for PVM */

#include "all.h"

/* Some compilers require at least one external reference */
extern int NProc;

#ifdef PVM
static int *PVM_tid; /* Task id array */
static int PVM_nproc,PVM_proc;
static int *PVM_dests;


/* Initialize n copies of the named process */
int
PVM_Init(char *name, int n)
{
 int i;
 int tid;
 tid=pvm_mytid();	/* Initializes PVM */

#ifdef T3D /* T3D has funky PVM setup!!! */
 /* Need to do some special setup since T3D auto spawns everything */
 pvm_tasks(0, &n, NULL);
 PVM_tid = (int *) New((SIZETYPE) n * sizeof(int));
 for(i=0; i<n; i++)
        PVM_tid[i] = i;
 PVM_nproc = n;
 PVM_proc = pvm_get_PE(tid);

#else

 tid = pvm_parent();	/* Check if we are children or the parent */

 if(tid < 0) /* Parent */
	{
	
	/* Set number of processors */
	PVM_nproc = n;

	/* Allocate the list */
	PVM_tid = (int *) New((SIZETYPE) n * sizeof(int));
	PVM_tid[0] = pvm_mytid();

	if(n<=1)
		{
		PVM_proc = 0;
		return(MPI_SUCCESS);
		}

	/* Spawn other tasks */
	if(pvm_spawn(name, NULL, 0, "", n-1, &PVM_tid[1]) < 0)
		return(MPI_FAILURE);

	/* Send number and tids of tasks to each */
	pvm_initsend(PvmDataDefault);
	pvm_pkint(&n,1,1);
	pvm_pkint(PVM_tid, n, 1);
	if(pvm_mcast(&PVM_tid[1], n-1, 0) < 0)
		return(MPI_FAILURE);
	}
 else
	{
	/* Children */
	if(pvm_recv(tid,0) < 0)
		return(MPI_FAILURE);
    /* Recieve number of tasks */
	pvm_upkint(&PVM_nproc, 1,1);

	/* Allocate list */
	PVM_tid = (int *) New((SIZETYPE) PVM_nproc * sizeof(int));

	/* Get the rest of the stuff */
	pvm_upkint(PVM_tid, PVM_nproc, 1);
	}

 /* Now we need to identify ourselves */
 tid = pvm_mytid();
#ifdef PVM_DIRECT
 /* Try out direct routing */
 pvm_setopt(PvmRoute, PvmRouteDirect);
#endif

 for(i=0; i<PVM_nproc; i++)
	{
	if(tid == PVM_tid[i])
		{
        PVM_proc = i;
		break;
		}
	}
#endif
 /* Create a dest array for multicasts */
 PVM_dests = (int *) New((SIZETYPE) PVM_nproc * sizeof(int));

 return(MPI_SUCCESS);
} 

int
PVM_Rank()
{
 return PVM_proc;
}

int
PVM_Nproc()
{
 return(PVM_nproc);
}

void
PVM_Abort(int errcode)
{
 char s[80];
 int i;

 if(!PVM_tid)
	PVM_Init("",1);

 sprintf(s, "Fatal error=%d rank=%d task=%xd - signalling PVM abort!\n",errcode, PVM_proc,
	PVM_tid[PVM_proc]);
 ReportError(s,FALSE);

 /* Kill everyone */
 for(i=0; i<PVM_nproc; i++)
 	if(i != PVM_proc)
		pvm_kill(PVM_tid[i]);
 /* Kill ourselves */
 pvm_exit();
}

/* Now capable of doing a multisend */
int
PVM_Send(void *buf, int count, MPI_Datatype data_type, int *dest, int ndest, int tag)
{
 int i;
 int self_send=FALSE;
#ifdef PVM_BUF
 int newbuf=-1, old_buf=-1;
#endif

 /* Begin send */
 for(i=0; i<ndest && i < PVM_nproc; i++)
	{
 	if(dest[i] < 0 || dest[i] >= PVM_nproc)
		{
		UserMessage("PVM_Send: Error - attempt to send to bad address '%d'\n",dest[i]);
		return(MPI_FAILURE);
		}
	if(dest[i] == PVM_proc)
		{
		self_send=TRUE;
		}
	/* Copy real address to destination array */
	PVM_dests[i] = PVM_tid[dest[i]];
	}

 /* Simply select appropriate PVM buffer ! */
 /* Note - count is overloaded -> actually the message id */
#define MY_BUFID count
 if(buf == NULL && MY_BUFID != pvm_getsbuf())
	{
	pvm_setsbuf(MY_BUFID);
	}
 else if(buf != NULL)
	{
 	/* Otherwise pack it up */
#ifdef PVM_BUF
	newbuf = pvm_mkbuf((data_type == MPI_BYTE)? PvmDataRaw: PvmDataDefault);
	if(newbuf < 0)
		return(MPI_FAILURE);
	old_buf = pvm_setsbuf(newbuf);
#else
 	if(pvm_initsend((data_type == MPI_BYTE)?PvmDataRaw:PvmDataDefault) < 0)
		return(MPI_FAILURE);
#endif
 	if(PVM_Pack(buf, count, data_type) < 0)
		return(MPI_FAILURE);
	}

 if(ndest == 1)
	{
 	if(pvm_send(PVM_dests[0], tag) < 0)
		return(MPI_FAILURE);
	}
 else
	{
	/* Broadcast to all !! */
 	if(pvm_mcast(PVM_dests, ndest, tag) < 0)
		return(MPI_FAILURE);
	if(self_send)
		if(pvm_send(PVM_tid[PVM_proc], tag) < 0)
			return(MPI_FAILURE);
	}
#ifdef PVM_BUF
	if(old_buf >= 0)
		pvm_setsbuf(old_buf);
 	if(newbuf >= 0)
		pvm_freebuf(newbuf);
#endif
	
 return(MPI_SUCCESS);
}

int
PVM_Receive(void *buf, int count, int data_type, int source,
	int tag, MPI_Status *status, int *buf_id)
{
 
 if(source >= PVM_nproc || source < MPI_ANY_SOURCE)
	{
	UserMessage("PVM_Receive: Attempt to receive on bad source=%d\n", source);
	return(MPI_FAILURE);
	}
 *buf_id=pvm_recv(PVM_SOURCE(source),tag);
 if(*buf_id < 0)
	return(MPI_FAILURE);

 PVM_GetStatus(*buf_id,count, data_type, source, status);

 if(buf)
 	return(PVM_Unpack(*buf_id,buf, status->count, data_type, source, status));

 return(MPI_SUCCESS);
}

int
PVM_Pack(void *buf, int count, MPI_Datatype data_type)
{
 int i;
 int ret;

 switch(data_type)
	{
        default:
	case MPI_BYTE:
	case MPI_UNSIGNED_CHAR:
    case MPI_CHAR:
		ret=pvm_pkbyte((char *)buf, count, 1);
		break;
	case MPI_INT:
		ret=pvm_pkint((int *)buf, count, 1);
		break;
 	case MPI_UNSIGNED:
		ret=pvm_pkuint((unsigned int *)buf, count, 1);
		break;
	case MPI_LONG:
		ret=pvm_pklong((long *) buf, count, 1);
		break;
	case MPI_UNSIGNED_LONG:
		ret=pvm_pkulong((unsigned long *) buf, count, 1);
		break;
	case MPI_UNSIGNED_SHORT:
		ret=pvm_pkushort((unsigned short *) buf, count, 1);
		break;
	case MPI_SHORT:
		ret=pvm_pkshort((short *)buf, count,1 );
		break;
	case MPI_FLOAT:
		ret=pvm_pkfloat((float *) buf, count, 1);
		break;
	case MPI_DOUBLE:
		ret=pvm_pkdouble((double *) buf, count, 1);
		break;
	}

 return((ret < 0)? MPI_FAILURE: MPI_SUCCESS);
}

int
PVM_Unpack(int buf_id,void *buf, int count, MPI_Datatype data_type, int source,
	MPI_Status *status)
{
  int i;
  int ret;

  switch(data_type)
	{
        default:
	case MPI_BYTE:
	case MPI_UNSIGNED_CHAR:
	case MPI_CHAR:
		ret=pvm_upkbyte((char *)buf, count, 1);
		break;
	case MPI_INT:
		ret=pvm_upkint((int *)buf, count, 1);
		break;
	case MPI_UNSIGNED:
		ret=pvm_upkuint((unsigned int *)buf, count, 1);
		break;
	case MPI_LONG:
		ret=pvm_upklong((long *) buf, count, 1);
		break;
	case MPI_UNSIGNED_LONG:
		ret=pvm_upkulong((unsigned long *) buf, count, 1);
		break;
	case MPI_UNSIGNED_SHORT:
		ret=pvm_upkushort((unsigned short *) buf, count, 1);
		break;
	case MPI_SHORT:
		ret=pvm_upkshort((short *)buf, count,1 );
		break;
	case MPI_FLOAT:
		ret=pvm_upkfloat((float *) buf, count, 1);
		break;
	case MPI_DOUBLE:
		ret=pvm_upkdouble((double *) buf, count, 1);
		break;
	}

 return((ret < 0)?MPI_FAILURE:MPI_SUCCESS);
}

void
PVM_GetStatus(int buf_id,  int count, MPI_Datatype datatype, int source, MPI_Status *status)
{
 int i;
 pvm_bufinfo(buf_id, &status->count, &status->tag, &status->source);

 /* get the id of the sender */
 if(source == MPI_ANY_SOURCE)
 	{
 	for(i=0; i<PVM_nproc; i++)
		if(status->source == PVM_tid[i])
			{
			status->source = i;
			break;
			}
	}
 else
	status->source = source;

 /* Determine the real count */
 status->count = MIN(count, status->count / PVM_SizeDataType(datatype));
}

int
PVM_Isend(void *buf, int count, MPI_Datatype data_type, int dest,
	int tag, MPI_Comm_request *request)
{
 /* Actually do the send */
 if(PVM_Send(buf,count, data_type, &dest,1, tag) < 0)
	return(MPI_FAILURE);
 request->status.source = dest;
 request->status.tag = tag;
 request->status.count = count;
 request->is_send = TRUE;
 return(MPI_SUCCESS);
}

int
PVM_Irecv(void *buf, int count, MPI_Datatype data_type, int source,
	int tag, MPI_Comm_request *request)
{
 request->status.source = source;
 request->status.tag = tag;
 request->status.count = count;
 request->is_send = FALSE;
 request->datatype = data_type;
 request->buf = buf;
 return(MPI_SUCCESS);
}

int
PVM_Wait(MPI_Comm_request *request, MPI_Status *status)
{
 int bufid;

 if(request->is_send)
	{
	*status = request->status;
    return(MPI_SUCCESS);
	}
 else
	return(PVM_Receive(request->buf, request->status.count, request->datatype,
		request->status.source, request->status.tag, status, &bufid));
}

int
PVM_Test(MPI_Comm_request *request, MPI_Status *status)
{
 int buf_id;
 if(request->is_send)
	{
	*status = request->status;
    return(TRUE);
	}
 else
	{
	if((buf_id=pvm_nrecv(PVM_SOURCE(request->status.source), request->status.tag)) <= 0)
        	return(FALSE);
	 PVM_Unpack(buf_id,request->buf, request->status.count, request->datatype,
		request->status.source, status);
	 return(TRUE);
	}
}

int
PVM_Probe(int source, int tag, int *flag, MPI_Status *status)
{
 int buf_id;

 buf_id = pvm_probe(PVM_SOURCE(source), tag);
 if(buf_id <=0)
	{
	*flag = FALSE;
	return(MPI_SUCCESS);
	}
 PVM_GetStatus(buf_id,  1, MPI_CHAR, source, status);
 *flag = TRUE;
 return(MPI_SUCCESS);
}

int
PVM_SizeDataType(MPI_Datatype datatype)
{
#if !defined(ALPHA) && !defined(CRAY) && !defined(T3D)
	return(_MPI_SizeDataType(datatype));
#else
switch(datatype)
	{
	default:
	case MPI_INT:
		return(4);
	case MPI_UNSIGNED:
		return(4);
	case MPI_SHORT:
    	return(sizeof(short));
	case MPI_UNSIGNED_SHORT:
		return(sizeof(unsigned short));
	case MPI_LONG:
		return(4);
	case MPI_UNSIGNED_LONG:
		return(4);
	case MPI_BYTE:
	case MPI_CHAR:
		return(sizeof(char));
	case MPI_DOUBLE:
		return(sizeof(double));
	case MPI_FLOAT:
		return(4);
	}
#endif
}

#endif /* PVM */