pm_separate_rect.c

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

#include <stdio.h>
#if HAVE_STDLIB_H
#include <stdlib.h>
#endif
#include <sys/time.h>
#include "pmandel.h"
#include "pm_genproc.h"

#ifndef DEBUG
#define DEBUG 0
#endif
#define MASTER_PROC 0
#define DEBUG_ASSIGNMENTS 0

void SplitRect(Flags *flags, rect r);

void SeparateRect_Master( graph, winspecs, flags )
MPE_XGraph graph;
Winspecs *winspecs;
Flags *flags;
{
  rect *assigList, recvRectBuf[2], tempRect;
  /* assigList - list of what rectangles have been assigned to each process */
  /* recvRectBuf - when a slave process tells the master that some rectangles
     need to be calculated, the rectangle definitions are temporarily stored
     here */

  /* The following 7 variables are used when only the master interacts
     with the X display. These variables are used in receiving the data
     from the slaves for display.
  */
  MPE_Point *pointData;
  int        dataSize;
  int        numPoints;
  int        block_type;
  rect       rectangle;
  int        color;
  int        getPoints; /* boolean, determines whether some computed points are
                           to be received from some rank.
                        */


  int inProgress, nidle, *idleList, splitPt, np, data,
      mesgTag, firstToEnqueue, procNum, i, randomPt;
  /* inProgress - number of rectangles currently under construction */
  /* nidle - number of idle processes */
  /* idleList - list of process ranks that are idle */
  /* splitPt - point at which to split the Mandelbrot set--can't have the
     border algorithm enclosing the whole set */
  /* np - number of processes */
  /* data - one int returned by slave processes; different uses */
  /* mesgTag - the tag of the received message */
  /* firstToEnqueue - when receiving a bunch of rectangles to enqueue, the
     master process may send some off to idle processes right away before
     queueing them.  firstToEnqueue points to the first one that is actually
     queued */
  /* procNum - rank of the process that sent the last message */
  /* randomPt - index+1 of the last item in the queue that has not been
                placed in random order */

  rect_queue rect_q;
  /* queue of rectangles to calculate */

  MPI_Status mesgStatus;

  struct timeval time;

  MPI_Comm_size( MPI_COMM_WORLD, &np );
  gettimeofday( &time, 0 );
    /* initialize the random number generator for the -randomize option */
  srand48( time.tv_sec * time.tv_usec );

  MPE_DESCRIBE_STATE( S_COMPUTE, E_COMPUTE,
		     "Compute", "blue:gray" );
  MPE_DESCRIBE_STATE( S_DRAW_BLOCK, E_DRAW_BLOCK,
		     "Draw block", "yellow:gray3" );
  MPE_DESCRIBE_STATE( S_DRAW_RECT, E_DRAW_RECT,
		     "Draw border", "green:light_gray" );
  MPE_DESCRIBE_STATE( S_WAIT_FOR_MESSAGE, E_WAIT_FOR_MESSAGE,
		     "Wait for message", "red:boxes" );
  MPE_DESCRIBE_STATE( S_DRAW_CHUNK, E_DRAW_CHUNK, "Draw Chunk",
		     "steelBlue:2x2" );

  assigList  = (rect *) malloc( (np) * sizeof( rect ) );
  idleList   = (int *)  malloc( (np) * sizeof( rect ) );

  nidle = inProgress = 0;
  Q_Create( &rect_q, flags->randomize );
    /* create the queue */

  if (flags->imin<0 && flags->imax>0) { /* might have to split the set */
    splitPt = winspecs->height +
      NUM2INT( NUM_DIV( NUM_MULT( flags->imin, INT2NUM( winspecs->height ) ),
				    NUM_SUB( flags->imax, flags->imin ) ) );
    RECT_ASSIGN( tempRect, 0, winspecs->width-1, 0, splitPt-1 );
    tempRect.length = RectBorderLen( &tempRect );
    Q_Enqueue( &rect_q, &tempRect );
    RECT_ASSIGN( tempRect, 0, winspecs->width-1, splitPt, winspecs->height-1 );
    tempRect.length = RectBorderLen( &tempRect );
    Q_Enqueue( &rect_q, &tempRect );
  } else {
    RECT_ASSIGN( tempRect, 0, winspecs->width-1, 0, winspecs->height-1 );
    tempRect.length = RectBorderLen( &tempRect );
    Q_Enqueue( &rect_q, &tempRect );
  }

  if (flags->no_remote_X) {
    int x,y;
    /* figure out how much data might be received from a process that
       has computed a block of the output and allocate space for it.
    */
    x = flags->breakout * flags->breakout;
    y = 2 * (winspecs->height + winspecs->width );
    dataSize = ((y>x) ? y : x );
    pointData = (MPE_Point *) malloc( dataSize * sizeof( MPE_Point) );
  }

#if DEBUG
  fprintf( debug_file, "Master starting up\n" );
  fflush( debug_file );
#endif

  while (inProgress || !IS_Q_EMPTY(rect_q) ) {
    /* while someone's working, and the q is !empty */

    MPE_LOG_EVENT( S_WAIT_FOR_MESSAGE, 0, 0 );
    MPI_Probe( MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &mesgStatus );
    procNum = mesgStatus.MPI_SOURCE;
    mesgTag = mesgStatus.MPI_TAG;
    MPE_LOG_RECEIVE( procNum, mesgTag, 0 );
    MPE_LOG_EVENT( E_WAIT_FOR_MESSAGE, 0, 0 );

#if DEBUG
    fprintf( debug_file, "Master receives %d from %d\n", mesgTag, procNum );
    fflush( debug_file );
#endif

    if (flags->no_remote_X) {
      getPoints = 1; /* if != 0, receive computed points from some process.
                        This is only used to distinguish between the
                        READY_TO_START and READY_FOR_MORE messages.
                      */
    }
    

    switch (mesgTag) {

    case READY_TO_START:
      inProgress++; /* NOTE: This falls through!!, no break; statement */
      if (flags->no_remote_X) {
        getPoints = 0;
      }

    case READY_FOR_MORE:
        MPI_Recv( 0, 0, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG,
		  MPI_COMM_WORLD, &mesgStatus );

      if (flags->no_remote_X && getPoints) {
        /* Receive the computed points from this rank.
           This will either be a specification for a rectangle in which all
           the points are the same color, or this will be a block of
           points (int x, int y, int color), i.e. MPE_Points.
         */
        MPI_Recv( &block_type, 1, MPI_INT, procNum, BLOCK_TYPE,
  	        MPI_COMM_WORLD, &mesgStatus );

        if (block_type == POINTS) {

          MPI_Recv( &numPoints, 1, MPI_INT, procNum, POINT_COUNT,
  	          MPI_COMM_WORLD, &mesgStatus );
          if (numPoints > dataSize) {
            fprintf( stderr,"master: INTERNAL ERROR: "
                    "too many data points to receive\n" );
            exit( 1 );
          }

          MPI_Recv( pointData, 3*numPoints, MPI_INT, procNum, POINT_DATA,
	            MPI_COMM_WORLD, &mesgStatus );

          MPE_Draw_points( graph, pointData, numPoints );
          MPE_Update( graph );

          if (flags->with_tracking_win) {
            int i;
            /* Now indicate which process computed this section. */
            MPI_Recv( &color, 1, MPI_INT, procNum, TRACKING_COLOR,
  	            MPI_COMM_WORLD, &mesgStatus );
            for (i=0; i<numPoints; i++) {
              pointData[i].c = color;
            }
	    MPE_Draw_points( tracking_win, pointData, numPoints );
            MPE_Update( tracking_win );
          }

        } else if (block_type == RECTANGLE) {

          MPI_Recv( &rectangle, 5, MPI_INT, procNum, RECT_SPEC,
  	          MPI_COMM_WORLD, &mesgStatus );
          MPI_Recv( &color, 1, MPI_INT, procNum, RECT_COLOR,
  	          MPI_COMM_WORLD, &mesgStatus );
          pointData->c = color; /* this is all that is used from pointData */

	  DrawBlock( graph, pointData, &rectangle );
          MPE_Update( graph );

          if (flags->with_tracking_win) {
            /* Now indicate which process computed this section. */
            MPI_Recv( &color, 1, MPI_INT, procNum, TRACKING_COLOR,
  	             MPI_COMM_WORLD, &mesgStatus );
            pointData[0].c = color; /* This is all that is referenced 
                                       from the pointData pointer in
                                       the DrawBlock routine. */
	    DrawBlock( tracking_win, pointData, &rectangle );
          }

        } else {

          fprintf( stderr,"master: INTERNAL ERROR: unknown block type\n" );
        }

        MPE_Update( graph );
      }

      if (IS_Q_EMPTY( rect_q)) {  /* if the queue is empty, */
	idleList[nidle++] = procNum;  /* remember this process was left idle */
	inProgress--;
      } else {
	Q_Dequeue( &rect_q, &tempRect );
#if DEBUG_ASSIGNMENTS
	fprintf( stderr, "about to assign (%d,%d - %d,%d) to %d\n",
		tempRect.l, tempRect.t,
		tempRect.r, tempRect.b, procNum );
	while (getchar()!='\n');
#endif
	MPI_Send( &tempRect, 1, rect_type,
		 procNum, ASSIGNMENT, MPI_COMM_WORLD );
	MPE_LOG_SEND( procNum, ASSIGNMENT, sizeof( rect ) );
	  /* otherwise, assign it the next in the queue */
	assigList[procNum] = tempRect;
	  /* remember which rect this process is on */
#if DEBUG
	fprintf( debug_file, "Master assigns (%d %d %d %d) to %d\n",
		assigList[procNum].l,
		assigList[procNum].r, assigList[procNum].t,
		assigList[procNum].b,
		procNum );
	fflush( debug_file );
#endif
      }
      break;

    case ADD2Q:
        /* slave is posting more rectangles to be queued */
      MPI_Recv( recvRectBuf, 2, rect_type, procNum,
	        ADD2Q, MPI_COMM_WORLD, &mesgStatus );
#if DEBUG
	fprintf( debug_file, "Master receives (%d %d %d %d)\n",
		(recvRectBuf[0]).l,
		(recvRectBuf[0]).r,
		(recvRectBuf[0]).t,
		(recvRectBuf[0]).b );
	fprintf( debug_file, "Master receives (%d %d %d %d)\n",
		(recvRectBuf[1]).l,
		(recvRectBuf[1]).r,
		(recvRectBuf[1]).t,
		(recvRectBuf[1]).b );
	fflush( debug_file );
#endif
        /* get rect definitions */
      firstToEnqueue = 0;
      while (nidle && firstToEnqueue < 2) {
	  /* if processes are idle, */	
	nidle--;
	assigList[idleList[nidle]] = recvRectBuf[firstToEnqueue];
	  /* remember which rect this process is on */
	MPI_Send( recvRectBuf + firstToEnqueue, 1, rect_type,
		  idleList[nidle], ASSIGNMENT, MPI_COMM_WORLD );
	MPE_LOG_SEND( idleList[nidle], ASSIGNMENT, sizeof( rect ) );
	  /* give them something to do */
#if DEBUG
	fprintf( debug_file, "Master assigns (%d %d %d %d) to %d\n",
		(recvRectBuf[firstToEnqueue]).l,
		(recvRectBuf[firstToEnqueue]).r,
		(recvRectBuf[firstToEnqueue]).t,
		(recvRectBuf[firstToEnqueue]).b, idleList[nidle] );
	fflush( debug_file );
#endif
	inProgress++; firstToEnqueue++;
      }
      for (; firstToEnqueue<2; firstToEnqueue++) {
	Q_Enqueue( &rect_q, recvRectBuf+firstToEnqueue );
#if DEBUG
	fprintf( debug_file, "Master queues (%d %d %d %d)\n",
		(recvRectBuf[firstToEnqueue]).l,
		(recvRectBuf[firstToEnqueue]).r,
		(recvRectBuf[firstToEnqueue]).t,
		(recvRectBuf[firstToEnqueue]).b );
	fflush( debug_file );
#endif
      }
      break;
    }
  }

#if DEBUG
  fprintf( debug_file, "All done\n" );
  fflush( debug_file );
#endif

  for (i=1; i<np; i++) {
      /* tell everyone to exit. */
    MPI_Send( rect_q.r, sizeof( rect ), MPI_BYTE, i, ALL_DONE, MPI_COMM_WORLD );
    MPE_LOG_SEND( i, ALL_DONE, sizeof( rect ) );
  }

  Q_Destroy( &rect_q );
  free( assigList );