source: gs2-extensions/parallel-building/trunk/src/src/mpiterrierfileindexer-src/mpiterrierfileindexer.cpp@ 26239

// Master-worker program to prepare, parallel index, and then merge a terrier
// based collection
//
// John Thompson
// 3rd August 2012

#include "mpi.h"

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

#include <fstream>
#include <iostream>
#include <sstream>
#include <string>
#include <vector>

using namespace std;

#define BUFFERSIZE 512

int
main ( int argc, char *argv [] )
{
  int numtasks, rank, rc;            // MPI variables
  unsigned long int seconds = 0;

  if (4 != argc )
  {
    fprintf(stderr,"Usage: mpiterrierfileindexer <path to gsdl> <path to terrier> <number of manifest files>\n");
    exit(-1);
  }

  char *gsdlhome_dir = argv[1];
  char *terrier_dir = argv[2];
  int total_number_of_manifest_files = atoi(argv[3]);

  // start MPI environment
  rc = MPI_Init(&argc,&argv);
  if (rc != MPI_SUCCESS)
  {
    fprintf(stderr, "Error starting MPI program. Terminating.\n");
    MPI_Abort(MPI_COMM_WORLD, rc);
  }

  // We'll handle errors ourselves
  MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_RETURN);

  // get MPI variables: number of processors and processor number
  MPI_Status stat;
  MPI_Comm_size(MPI_COMM_WORLD, &numtasks);
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);

  // Get processor name too - important when it could be anywhere in a cluster
  int name_length;
  char processor_name[MPI_MAX_PROCESSOR_NAME];
  MPI_Get_processor_name(processor_name, &name_length);

  // master node processing
  if (rank == 0)
  {
    seconds = time(NULL);
    fprintf(stderr, "[M:%lu] Starting on %s\n", seconds, processor_name);
    char incoming[BUFFERSIZE];          // buffer for acknowledgments
    char buffer[BUFFERSIZE];         // buffer to send tasks
    MPI_Request request[BUFFERSIZE];    // request monitor for all tasks
    MPI_Status status[BUFFERSIZE];      // status monitor for all tasks
    int actualTasks = 0;         // number of processors running
    int manifest_counter = 0;

    // set initial status of all processors to idle
    fprintf(stderr, "[M] Initializing processor state\n");
    for ( int j=0; j<BUFFERSIZE; j++ )
    {
      incoming[j] = ' ';
    }

    // scan through contents of file listing
    fprintf(stderr, "[M] Running work process for each manifest file\n");
    while (manifest_counter < total_number_of_manifest_files)
    {
      // search for idle processor
      int dest=0;
      int found = 0;
      fprintf(stderr, "[M] Searching for idle processor\n");
      while ((dest<(numtasks-1)) && (found == 0))
      {
        if (incoming[dest] == ' ')
        {
          found = 1;
        }
        else
        {
          dest++;
        }
      }

      // if no idle processor, wait for one to become idle
      if (found == 0)
      {
        fprintf(stderr, "[M] Waiting for processor to become idle\n");
        MPI_Waitany (numtasks-1, request, &dest, status);
      }

      // Send manifest number (as 0 padded 3 digit string) to the worker
      fprintf(stderr, "[M] Writing manifest number as instruction to worker\n");
      // Jiggerypokery to get around weird compiler error: cannot pass
      // objects of non-POD type ‘struct std::string’ through ‘...’; call
      // will abort at runtime
      sprintf(buffer, "%03d", manifest_counter);

      // mark processors as busy
      incoming[dest] = 'B';
      // send out the job to the processor
      fprintf(stderr, "[M] Asking worker %d to start\n", (dest + 1));
      MPI_Send (&buffer, strlen (buffer)+1, MPI_CHAR, dest+1, 1, MPI_COMM_WORLD);
      // wait for a done acknowledgement
      MPI_Irecv (&incoming[dest], 1, MPI_CHAR, dest+1, 1, MPI_COMM_WORLD, &request[dest]);
      fprintf(stderr, "[M] Worker %d replied that it has started\n", (dest + 1));
      // update counter of actual tasks
      if (dest > actualTasks)
      {
        actualTasks = dest;
        fprintf(stderr, "[M] Increased the number of running workers to: %d\n", actualTasks);
      }

      // onto the next manifest file
      manifest_counter++;
    }

    // wait until all outstanding tasks are completed
    fprintf(stderr, "[M] Waiting for all outstanding tasks to complete\n");
    int dest;
    for ( int k=0; k<actualTasks; k++ )
    {
      MPI_Waitany (actualTasks, request, &dest, status);
    }

    // send message to end all processing engines
    fprintf(stderr,"[M] Master asking children to exit\n");
    char endstr[5] = "end";
    for ( int i=1; i<numtasks; i++ )
    {
      MPI_Send (endstr, 4, MPI_CHAR, i, 1, MPI_COMM_WORLD);
    }

    seconds = time(NULL);
    fprintf(stderr, "[M:%lu] Exiting\n", seconds);
  }

  // worker node processing
  else
  {
    seconds = time(NULL);
    fprintf(stderr, "[W%d:%lu] Starting on %s\n", rank, seconds, processor_name);

    // Check to see if GSDLHOME exists in the environment (it will on multicore
    // computer, but won't on compute nodes in a cluster). It will be NULL if
    // source setup.bash hasn't been run (on this computer).
    const char * gsdlhometest = getenv("GSDLHOME");

    char incoming[BUFFERSIZE];
    int counter = 0;
    do
    {
      // wait for instruction from master
      int resval = MPI_Recv (&incoming, BUFFERSIZE, MPI_CHAR, 0, 1, MPI_COMM_WORLD, &stat);
      if (resval != MPI_SUCCESS)
      {
        fprintf(stderr, "[W%d] Error when recieving message from master... terminating (%d).\n", rank, resval);
        MPI_Abort(MPI_COMM_WORLD, rc);
      }
      counter++;
      if (strcmp (incoming, "end") != 0)
      {
        // process a received job
        seconds = time(NULL);
        fprintf(stderr, "[W%d:%lu] Processing: %s\n", rank, seconds, incoming, counter);

        // create DSpace filter-media command
        // - incoming now contains the identifier of the item to filter
        char command[2048];
        if (gsdlhometest != NULL)
        {
          sprintf (command, "%s/bin/anyclass.sh org.terrier.applications.FileIndexer -index -path %s/var/manifest-%s.spec -prefix %s > /tmp/terrier-index-P%d-C%d.log 2>&1", terrier_dir, terrier_dir, incoming, incoming, rank, counter);
        }
        else
        {
          sprintf (command, "bash -c \"cd %s && source setup.bash > /dev/null && %s/bin/anyclass.sh org.terrier.applications.FileIndexer -index -path %s/var/manifest-%s.spec -prefix %s > /tmp/terrier-index-P%d-C%d.log 2>&1\"", gsdlhome_dir, terrier_dir, terrier_dir, incoming, incoming, rank, counter);
        }
        fprintf(stderr, "[W%d] system('%s')\n", rank, command);
        // invoke dspace
        system(command);

        // send completed message
        char line = ' ';
        MPI_Send (&line, 1, MPI_CHAR, 0, 1, MPI_COMM_WORLD);
        seconds = time(NULL);
        fprintf(stderr, "[W%d:%lu] Process complete\n", rank, seconds);
      }
    }
    while (strcmp (incoming, "end") != 0);
    // stop when "end" instruction is received
    seconds = time(NULL);
    fprintf(stderr, "[W%d:%lu] Exiting\n", rank, seconds);
  }

  // clean up MPI environment
  if (rank == 0)
  {
    fprintf(stderr,"[M] Finalizing...\n");
  }
  else
  {
    fprintf(stderr,"[W%d] Finalizing...\n", rank);
  }
  MPI_Finalize();
}