Changeset 25839

Timestamp:

2012-06-28T09:17:21+12:00 (11 years ago)

Author:

jmt12

Message:

Adding more timing to threads

File:

• gs2-extensions/parallel-building/trunk/src/src/mpiimport-src/mpiimport.cpp (modified) (19 diffs)

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

@@ -18 +18 @@
 #include <stdio.h>
 #include <stdlib.h>
+#include <time.h>
 
 #include <fstream>
@@ -34 +35 @@
 {
   int numtasks, rank, rc;            // MPI variables
+  unsigned long int seconds = 0;
 
   if (5 != argc && argc != 6 )
   {
-    cerr << "Usage: " << argv[0] << " filelist epoch gsdlhome collection [site]" << endl;
+    fprintf(stderr, "Usage: mpiimport filelist epoch gsdlhome collection [site]\n");
     exit(-1);
   }
@@ -46 +48 @@
   char *gsdlhomedir = argv[3];      // location of import script
   char *collection = argv[4];        // Greenstone collection
-  char *site = "";
+  char *site = NULL;
   if (argc == 6)
   {
@@ -56 +58 @@
   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);
   }
@@ -68 +70 @@
   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
@@ -86 +94 @@
 
     // 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++ )
     {
@@ -94 +102 @@
     // scan through contents of file listing
     int manifest_file_count = 1;
-    cerr << " - processing contents of filelist.txt" << endl;
+    fprintf(stderr, "[M] Processing contents of filelist.txt\n");
     while (!infile.eof ())
     {
       // get a filename
       getline (infile, line);
+
+      if ( line.length() == 0 && !infile.eof() )
+      {
+        fprintf(stderr, "[M] Warning! Read empty string from filelist: %s", filelist);
+      }
+
       if (line.length() > 0)
       {
@@ -104 +118 @@
         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))
         {
@@ -120 +134 @@
         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);
         }
 
         // construct manifest filename
-        cerr << " - creating manifest file: number " << manifest_file_count << endl;
+        fprintf(stderr, "[M] Creating manifest file: %d\n", manifest_file_count);
     stringstream manifestfilename_strstr;
     manifestfilename_strstr << gsdlhomedir << "/collect/" << collection << "/tmp/manifest." << manifest_file_count << ".xml";
@@ -138 +152 @@
           if (manifestfile.fail())
           {
-            cerr << "Fatal Error! Failed to open temporary manifest file for writing: " << manifestfile << endl;
+            fprintf(stderr, "[M] Fatal Error! Failed to open temporary manifest file for writing: %s\n", manifestfilename);
           }
         }
         else
         {
-          cerr << "Fatal Error! Failed to open temporary manifest file for writing: " << manifestfile << endl;
+          fprintf(stderr, "[M] Fatal Error! Failed to open temporary manifest file for writing: %s\n", manifestfilename);
         }
         manifestfile << "<Manifest><Index>" << endl;
@@ -165 +179 @@
         manifestfile.close ();
 
-        cerr << " - manifest file complete" << endl;
+        fprintf(stderr, "[M0] Manifest file complete\n");
 
         // Store manifest filepath so we can remove it later
@@ -171 +185 @@
 
         // Send the manifest filename as the instruction
-        cerr << " - writing manifest filename as instruction to worker" << endl;
+        fprintf(stderr, "[M0] Writing manifest filename as instruction to worker\n");
         sprintf(buffer, "%s", manifestfilename);
 
@@ -177 +191 @@
         incoming[dest] = 'B';
         // send out the job to the processor
-        cerr << " - asking worker to start" << endl;
+        fprintf(stderr, "[M0] 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, "[M0] 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, "[M0] Increased the number of running workers to: %d\n", actualTasks);
         }
     // increase number of manifest files processed
@@ -196 +210 @@
 
     // wait until all outstanding tasks are completed
-    cerr << " - waiting for all outstanding tasks to complete" << endl;
+    fprintf(stderr, "[M0] Waiting for all outstanding tasks to complete\n");
     int dest;
     for ( int k=0; k<actualTasks; k++ )
@@ -204 +218 @@
 
     // send message to end all processing engines
-    cerr << " * Master asking children to exit" << endl;
+    fprintf(stderr, "[M0] Master asking children to exit\n");
     char endstr[5] = "end";
     for ( int i=1; i<numtasks; i++ )
@@ -212 +226 @@
 
     // Free up manifest files
-    cerr << " - freeing shared memory used by manifest files" << endl;
+    fprintf(stderr, "[M0] Freeing shared memory used by manifest files\n");
     for ( int i = 0; i < manifest_files.size(); i++)
     {
       free(manifest_files[i]);
     }
-    cerr << " * Master Exiting" << endl;
-  }
-  // slave node processing
+    seconds = time(NULL);
+    fprintf(stderr, "[M0:%lu] Master will exit when workers complete\n", seconds);
+  }
+  // worker 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
     {
@@ -233 +253 @@
       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);
       }
@@ -240 +260 @@
       {
         // 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 Greenstone import command
         // - incoming now contains the name of the manifest file to process
         char command[2048];
-
 #ifdef HARDAFFINITY
         int cpu = rank - 1;
-        cerr << "Setting affinity for worker " << rank << " to cpu " << cpu << endl;
-        if (site != "")
-        {
-          sprintf (command, "taskset -c %d %s/bin/script/import.pl -verbosity 0 -keepold -manifest %s -site %s %s", cpu, gsdlhomedir, incoming, site, collection);
+        fprintf(stderr, "[W%d] Affinity fixed to CPU %d", rank, cpu);
+        char affinity[16];
+        sprintf(affinity, "taskset -c %d", cpu);
+#else
+        char affinity[16] = "";
+#endif
+        if (site != NULL && strlen(site) > 0)
+        {
+          if (gsdlhometest != NULL)
+          {
+            sprintf(command, "%s import.pl -keepold -manifest %s -site %s %s > /tmp/import-W%d-%d.log 2>&1", affinity, incoming, site, collection, rank, counter);
+          }
+          else
+          {
+            sprintf(command, "bash -c \"cd %s && source setup.bash > /dev/null && %s import.pl -verbosity 0 -keepold -manifest %s -site %s %s > /tmp/import-W%d-%d.log 2>&1", gsdlhomedir, affinity, incoming, site, collection, rank, counter);
+          }
+        }
+        else if (gsdlhometest != NULL)
+        {
+          sprintf(command, "%s import.pl -keepold -manifest %s %s > /tmp/import-W%d-%d.log 2>&1", affinity, incoming, collection, rank, counter);
         }
         else
         {
-          sprintf (command, "taskset -c %d %s/bin/script/import.pl -verbosity 0 -keepold -manifest %s %s", cpu, gsdlhomedir, incoming, collection);
-        }
-#else
-        if (site != "")
-        {
-          sprintf (command, "%s/bin/script/import.pl -verbosity 0 -keepold -manifest %s -site %s %s", gsdlhomedir, incoming, site, collection);
-        }
-        else
-        {
-          //sprintf (command, "%s/bin/script/import.pl -verbosity 0 -keepold -manifest %s %s", gsdlhomedir, incoming, collection);
-          sprintf (command, "%s/bin/script/import.pl -keepold -manifest %s %s > %s/collect/%s/logs/import-p%d-%d.log 2>&1", gsdlhomedir, incoming, collection, gsdlhomedir, collection, rank, counter);
-        }
-#endif
-
-        cerr << "**** cmd = " << command << endl;
+          sprintf(command, "bash -c \"cd %s && source setup.bash > /dev/null && %s import.pl -keepold -manifest %s %s > /tmp/import-W%d-%d.log 2>&1\"", affinity, gsdlhomedir, incoming, collection, rank, counter);
+        }
+
+        fprintf(stderr, "[W%d] system('%s')\n", rank, command);
 
         // invoke Greenstone import with manifest file
         system (command);
-        cerr << "**** complete" << endl;
+        seconds = time(NULL);
+        fprintf(stderr, "[W%d:%lu] Process complete\n", rank, seconds);
         char line = ' ';
         // send completed message
@@ -281 +306 @@
     while (strcmp (incoming, "end") != 0);
     // stop when "end" instruction is received
-    cerr << "Worker Exiting" << endl;
-  }
-
-  // clean up MPI environment
-  cerr << "Finalizing..." << endl;
+    seconds = time(NULL);
+    fprintf(stderr, "[W%d:%lu] Worker exiting\n", rank, seconds);
+  }
+  ///err << "Finalizing..." << endl;
   MPI_Finalize();
 }