m.c

基于大型机的秉性运算程序实现的是用多个agent对多任务的调度。程序实现的还不够成熟

#include "mpi.h"
#include <stdio.h>
#include <malloc.h>
#define L 4
#define TOTAL_TASK_NUM 10 
typedef struct request
{
	int type; //type of task
	int order; //the task serials
	int nT;    //the time that evry processor will run
	struct request* next;
} req;
void tAgent(int*, int task[10][5] );
void mAgent(int*,int task[10][5]);
void cAgent(int);
void wAgent();
req * insert(req*, req* );
req * creat(int,int (*task)[5]); //creat task table
int main(int argc, char*argv[])
{
	int rank,size;
	int* level;
	MPI_Init(&argc,&argv);
	MPI_Comm_rank(MPI_COMM_WORLD,&rank);
	if(rank == 0){
		int task[10][5];
		level = (int*)malloc(sizeof(int)*L);
		tAgent(level,task);
		mAgent(level,task);
	}
	else{
		int pStatus=0; //the status of the process; 
		               // if status=0,idle;if status=1,working;if status=2,over;
		wAgent();
		}
	MPI_Finalize();
	return 0;
}
void tAgent(int *level,int task[10][5])
{
	int i,j;
	int tUnit;
	tUnit = 10000;
	memset(task,0,sizeof(int)*50);
	for(i=0;i<2;i++)
		level[i] = 3;
	level[3] = 1;
//the 1-D is level of the tasks;
	for(i=0;i<10;i++)
		task[i][0] = 0;//used for noting the status about which the task is done.
//the 2-D is the type of the tasks;
	task[2][1] = 1;
	task[7][1] = 2;
//the 3-D~5-D is the necessary time of the task
	task[0][2] = 10 * tUnit;
	task[1][2] = 8 * tUnit;
	task[2][2] = 8 * tUnit;
	task[2][3] = 4 * tUnit;
	task[2][4] = 4 * tUnit;
	task[3][2] = 12 * tUnit;
	task[4][2] = 20 * tUnit;
	task[5][2] = 2 * tUnit;
	task[6][2] = 7 * tUnit;
	task[7][2] = 2 * tUnit;
	task[7][3] = 2 * tUnit;
	task[7][4] = 2 * tUnit;
	task[8][2] = 21 * tUnit;
	task[9][2] = 10 * tUnit;

}
void mAgent(int *level,int task[10][5])
{
	int i,j,nTask0,n,k,numOfTask,overFlag;
	int *pFlag;
	int rank,size,des,ranKFrom;
	int pStatus,pTaskType,subTaskType,taskOrder,nT;
	int counter,sendData[3],recvData[2];
	req *head,*q,*newTask;
	MPI_Comm_rank(MPI_COMM_WORLD,&rank);
	MPI_Comm_size(MPI_COMM_WORLD,&size);
	MPI_Status status;
	MPI_Request handle;
	pFlag = (int*)malloc(sizeof(int)*(size-1));
	for (i=0;i<(size-1);i++)
		pFlag[i] = 0; //if rank i is idle,the pFlag[i] will be 0;else means it working and noted 1
	n = 0;
	nTask0 = 0;
	overFlag = 0;
	pStatus = 1;
	for (j=0;j<L;j++) {
		numOfTask = level[j];
		//taskTable = (req*)malloc(sizeof(req)*numOfTask);
		head = creat(numOfTask,&task[nTask0]); //creat task table for level i
        	q = head;
		for (i=0;i<(size-1);i++) {
			counter++;
			if(q != NULL){
				if(pFlag[i]==0){
					des = i+1;
					pTaskType = q->type;
					n = q->order;
					nT = q->nT;
					sendData[0] = pTaskType;
					sendData[1] = n;
					sendData[3] = nT;
		           		if(pTaskType == 2)
						sendData[0] = pTaskType * 100;
					MPI_Send(&pStatus,1,MPI_INT,des,10,MPI_COMM_WORLD);
					MPI_Recv(0,0,MPI_INT,des,0,MPI_COMM_WORLD,&status);
					MPI_Send(sendData,3,MPI_INT,des,11,MPI_COMM_WORLD);
					pFlag[i] = 0;
					q = q->next;
					if (pTaskType == 2) {
						newTask->type = pTaskType * 100;
						newTask->order = n;
						newTask->nT = task[n][3];
						q = insert(q,newTask);	
					}
				}
			    else continue;
			}
			else break;
		}
		while(1){
			MPI_Recv(recvData,2,MPI_INT,MPI_ANY_SOURCE,100,MPI_COMM_WORLD,&status);
			ranKFrom = status.MPI_SOURCE;
			pTaskType = recvData[0]; //the type of the task
			n = recvData[1];         //the taskorder
			task[n][0]++;
			if(pTaskType == 1){
				for(k=0;k<2;k++)
				newTask->type =pTaskType*100+k ;
				newTask->order = n;
				newTask->nT = 3 + k;
				q = insert(q,newTask);
			}
			if (pTaskType == 200) {
				if(task[n][0] == 2){
					newTask->type = 200;
					newTask->order = n;
					newTask->nT = task[n][4];
					q = insert(q,newTask);
				}
			}
			if(q == NULL){
				for (k=0;k<TOTAL_TASK_NUM;k++) {
					if(task[k][1]==0){
						if(task[k][0] == 1)
							continue;
						else
							break;
					}
					else if(task[k][0]==3)
						continue;
					else
						break;
				}
				if(k==TOTAL_TASK_NUM){
					if(j==L)
						overFlag = 1;
					break;
				}
			}
			else
				break;
		}
	nTask0 += numOfTask;
	}
		

	//MPI_Bcast(&overFlag,1,MPI_INT,0,MPI_COMM_WORLD);//run this by all processors	
}
void wAgent()
{
	int pStatus,taskType,overFlag,nT;
	int subTask[3];
	MPI_Status status;
	while(1){
		pStatus = 0;
		MPI_Recv(&pStatus,1,MPI_INT,0,10,MPI_COMM_WORLD,&status);
		if (pStatus == 1) {
			MPI_Send(0,0,MPI_INT,0,0,MPI_COMM_WORLD);
			MPI_Recv(subTask,3,MPI_INT,0,11,MPI_COMM_WORLD,&status);
			nT = subTask[2];
			cAgent(nT);
			MPI_Send(subTask,2,MPI_INT,0,100,MPI_COMM_WORLD);
			//MPI_Recv(&overFlag,1,MPI_INT,0,0,MPI_COMM_WORLD,&status);
			//if(overFlag == 1)
			//	break;
		}
	}	
}
void cAgent(int nT)
{
	int i=0;
	while (i<nT) {
		i++;
	}

}
req * creat(int numOfTask,int (*task)[5]) //creat task table
{
	int i,num;
	req * head,*table;
	table = (req*)malloc(sizeof(req)*numOfTask);
	head = &table[0];
	num = numOfTask - 1;
	for(i=0;i<num;i++){
		table[i].type = task[i][1];
		table[i].order = i;
		table[i].nT = task[i][2];
		table[i].next = &table[i+1];
	}
	table[num].type = task[num][1];
	table[num].order = num;
	table[num].nT = task[i][3];
 	table[num].next = NULL;
	return head;
}
req * insert(req *head, req * rq )
{
	req * p0,*p1;
	p1 = head;
	p0 = rq;
	if(head == NULL){head = p0; p0->next = NULL;}
	else{head = p0;p0->next =p1;}
	return head;
}
int check(int *pFlag,int size)
{
	int i,n;
	n = size-1;
	for(i=0;i<n;i++)
		if(pFlag[n] == 1)
			return 0;
		else 
			continue;
	return 1;
}