Hi i have a strange problem.
My processes are like this.
Each process has a thread to recv messages(MPI non blocking recv). The thread loops infinitely.
The main thread searches for a value in its array of known value. If not found it tries to get it from other processes.It sends req for the same using ; MPI send call.(non blocking MPI send).
I am using 6 such processes.
My problem is i get proper results some time, sometime it just hangs.
Is it like if many messages are send to a process by others at a time, the receiver process cant handle? it misses something?
Is it problems with threads? I use Posix threads.
I am attaching the program...though other functions are not that important for others.
Any idea, or suggestion regarding implementation of concurrent threads is welcome.
Thanks in advance
Regards,
Imran
#include<iostream>
#include "mpi.h"
#include<pthread.h>
#include<signal.h>
#include<unistd.h>
#define CACHESIZE 10
#define HOPCOUNT 5
#define RESULTLIMIT 30
#define MAXRESOURCES 30
#define DISCARDTAG 9999
using namespace std;
using namespace MPI;
int search(int);
void* listen(void*);
int get_best_neighbor(int);
int getindex(int);
void sig_handler(int);
void add_cache(int,int);
int check_cache(int);
int me,nbrcnt,res_dist_size,reqcnt,fwdcnt,respcnt,cachecnt;
int *nbrs,*expbase,*myint;
int cache[CACHESIZE][2];
MPI::Graphcomm gcomm;
pthread_t tid;
int main(int argc, char**argv)
{
int i,wsize,membershipkey;
int result;
const int index[6]={1,4,6,9,10,12};
const int edges[12]={1,0,2,5,1,3,2,4,5,3,1,3};
int nnodes;
reqcnt=fwdcnt=respcnt=cachecnt=0;
MPI::Init_thread(argc,argv,2);
me=MPI::COMM_WORLD.Get_rank();
wsize=MPI::COMM_WORLD.Get_size();
nnodes=6;
gcomm=MPI::COMM_WORLD.Create_graph(nnodes,index,edges,false);
nbrcnt=gcomm.Get_neighbors_count(me);
nbrs=new int[nbrcnt];
expbase=new int[nbrcnt];
gcomm.Get_neighbors(me,nbrcnt,nbrs);
for(int i=0;i<nbrcnt;i++)
{
expbase[i]=0;
}
res_dist_size=MAXRESOURCES/wsize;
myint=new int[res_dist_size];
for(int i=res_dist_size*me;i<res_dist_size*(me+1);i++)
{
myint[i-(res_dist_size)*me]=i;
}
cout<<endl;
gcomm.Barrier();
if(pthread_create(&tid,0,listen,(void*)&me)<;0)
{
cout<<"Thread cant be created"<<endl;
}
{
int randnum;
switch(me)
{
case 0:
randnum=4;
break;
case 1:
randnum=10;
break;
case 2:
randnum=8;
break;
case 3:
randnum=10;
break;
case 4:
randnum=15;
break;
case 5:
randnum=8;
break;
default :
randnum=27;
}
int sendbuf[2];
reqcnt++;
cout<<"P="<<me<<" RN="<<randnum<<endl;
int
resrc_cnt=search(randnum);
if(resrc_cnt>0)
{
respcnt++;
cout<<"Yahoo! P="<<;me<<" A="<<;randnum<;<" Respcnt="<<respcnt<<endl;
}
else
{
sendbuf[0]=randnum;
sendbuf[1]=me;
int bestneighbor=get_best_neighbor(me);
add_cache(randnum,bestneighbor);
MPI::Request request;
MPI::Status status;
request=gcomm.Isend(&sendbuf,2,MPI::INT,bestneighbor,HOPCOUNT);
fwdcnt++;
cout<<"P="<<me<<" NA="<<randnum<<" RF-->"<<bestneighbor<;<endl;
request.Wait(status);
}
}
int ret;
int *retval=&ret;
pthread_join(tid,(void**)&retval);
gcomm.Free(); |