Changeset 25944 for gs2-extensions/parallel-building/trunk/src/src/mpidspacemediafilter-src/mpidspacemediafilter.cpp

Timestamp:

2012-07-13T12:06:06+12:00 (12 years ago)

Author:

jmt12

Message:

More debug comments, fixed memory leak - still hanging unfortunately

File:

• gs2-extensions/parallel-building/trunk/src/src/mpidspacemediafilter-src/mpidspacemediafilter.cpp (modified) (13 diffs)

gs2-extensions/parallel-building/trunk/src/src/mpidspacemediafilter-src/mpidspacemediafilter.cpp

@@ -24 +24 @@
 {
   int numtasks, rank, rc;            // MPI variables
-
-  if (3 != argc )
-  {
-    cerr << "Usage: " << argv[0] << " dspacehome filelist" << endl;
+  unsigned long int seconds = 0;
+
+  if (4 != argc )
+  {
+    fprintf(stderr,"Usage: mpidspacemediafilter gsdlhome dspacehome filelist\n");
     exit(-1);
   }
 
-  char *dspacehomedir = argv[1];      // location of import script
-  char *filelist = argv[2];
+  char *gsdlhomedir = argv[1];
+  char *dspacehomedir = argv[2];      // location of import script
+  char *filelist = argv[3];
 
   // start MPI environment
@@ -38 +40 @@
   if (rc != MPI_SUCCESS)
   {
-    printf ("Error starting MPI program. Terminating.\n");
+    fprintf(stderr, "Error starting MPI program. Terminating.\n");
     MPI_Abort(MPI_COMM_WORLD, rc);
   }
@@ -50 +52 @@
   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)
   {
-    cerr << " * Master Starting" << endl;
+    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
@@ -66 +74 @@
 
     // set initial status of all processors to idle
-    cerr << " - initializing processor state" << endl;
+    fprintf(stderr, "[M] Initializing processor state\n");
     for ( int j=0; j<BUFFERSIZE; j++ )
     {
@@ -73 +81 @@
 
     // scan through contents of file listing
-    cerr << " - processing contents of filelist.txt" << endl;
+    fprintf(stderr, "[M] Processing contents of filelist.txt\n");
     while (!infile.eof ())
     {
@@ -83 +91 @@
         int dest=0;
         int found = 0;
-        cerr << " - searching for idle processor" << endl;
+        fprintf(stderr, "[M] Searching for idle processor\n");
         while ((dest<(numtasks-1)) && (found == 0))
         {
@@ -99 +107 @@
         if (found == 0)
         {
-          cerr << " - waiting for processor to become idle" << endl;
+          fprintf(stderr, "[M] Waiting for processor to become idle\n");
           MPI_Waitany (numtasks-1, request, &dest, status);
         }
@@ -105 +113 @@
         // No need for manifests - just send the identifier of the item
         // to filter to the worker
-        cerr << " - writing dspace identifier as instruction to worker" << endl;
+        fprintf(stderr, "[M] Writing dspace identifier 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
         char *line = new char [line_str.size() + 1];
         strcpy(line, line_str.c_str());
         sprintf(buffer, "%s", line);
+        delete [] line;
 
         // mark processors as busy
         incoming[dest] = 'B';
         // send out the job to the processor
-        cerr << " - asking worker to start" << endl;
+        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]);
-        cerr << " - worker replied that it has started" << endl;
+        fprintf(stderr, "[M] Worker %d replied that it has started\n", (dest + 1));
         // update counter of actual tasks
         if (dest > actualTasks)
         {
           actualTasks = dest;
-          cerr << " - increased the number of running workers to: " << actualTasks << endl;
+          fprintf(stderr, "[M] Increased the number of running workers to: %d\n", actualTasks);
         }
       }
@@ -130 +142 @@
 
     // wait until all outstanding tasks are completed
-    cerr << " - waiting for all outstanding tasks to complete" << endl;
+    fprintf(stderr, "[M] Waiting for all outstanding tasks to complete\n");
     int dest;
     for ( int k=0; k<actualTasks; k++ )
@@ -138 +150 @@
 
     // send message to end all processing engines
-    cerr << " * Master asking children to exit" << endl;
+    fprintf(stderr,"[M] Master asking children to exit\n");
     char endstr[5] = "end";
     for ( int i=1; i<numtasks; i++ )
@@ -145 +157 @@
     }
 
+    seconds = time(NULL);
+    fprintf(stderr, "[M:%lu] Master will exit when workers complete\n", seconds);
   }
   // slave node processing
   else
   {
-    ///out << "Worker Starting" << endl;
+    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
     {
@@ -160 +179 @@
       if (resval != MPI_SUCCESS)
       {
-        printf ("Error when recieving message from master [%d]... Terminating.\n", resval);
+        fprintf(stderr, "[W%d] Error when recieving message from master... terminating (%d).\n", rank, resval);
         MPI_Abort(MPI_COMM_WORLD, rc);
       }
@@ -167 +186 @@
       {
         // process a received job
-        cerr << "Worker Processing [" << rank << ":" << incoming << "]" << endl;
+        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];
-
-        sprintf (command, "%s/bin/dspace filter-media -f -i \"%s\" > /tmp/dspace_media_filter-P%d-C%d.log 2>&1", dspacehomedir, incoming, rank, counter);
-
-        cerr << "**** cmd = " << command << endl;
-
+        if (gsdlhometest != NULL)
+        {
+          sprintf (command, "%s/bin/dspace filter-media -f -i \"%s\" > /tmp/dspace_media_filter-P%d-C%d.log 2>&1", dspacehomedir, incoming, rank, counter);
+        }
+        else
+        {
+          sprintf (command, "bash -c \"cd %s && source setup.bash > /dev/null && %s/bin/dspace filter-media -f -i %s > /tmp/dspace_media_filter-P%d-C%d.log 2>&1\"", gsdlhomedir, dspacehomedir, incoming, rank, counter);
+        }
+        fprintf(stderr, "[W%d] system('%s')\n", rank, command);
         // invoke dspace
         system (command);
-        cerr << "**** complete" << endl;
+
+        // send completed message
         char line = ' ';
-        // send completed message
         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
-    cerr << "Worker Exiting" << endl;
+    seconds = time(NULL);
+    fprintf(stderr, "[W%d:%lu] Worker exiting\n", rank, seconds);
   }
 
   // clean up MPI environment
-  cerr << "Finalizing..." << endl;
+  if (rank == 0)
+  {
+    fprintf(stderr,"[M] Finalizing...\n");
+  }
+  else
+  {
+    fprintf(stderr,"[W%d] Finalizing...\n", rank);
+  }
   MPI_Finalize();
 }