Changeset 27654 for gs2-extensions/parallel-building/trunk

Timestamp:

2013-06-18T10:59:52+12:00 (11 years ago)

Author:

jmt12

Message:

Add the ability to stagger the starting of Mappers by placing a 'delay.me' file in the tmp directory of the compute node to delay. They will then be initially delayed by compute node number * 100 seconds

Location:

gs2-extensions/parallel-building/trunk/src

Files:

• bin/script/hadoop_import.pl (modified) (10 diffs)
• src/java/org/nzdl/gsdl/HadoopGreenstoneIngest.java (modified) (2 diffs)

gs2-extensions/parallel-building/trunk/src/bin/script/hadoop_import.pl

@@ -22 +22 @@
 my $debug = 0;
 my $dry_run = 0;
+my $stagger = 0;
 my $flush_diskcache = 0;
 my $use_nfs = 0;
@@ -44 +45 @@
 else
 {
-  print STDERR "usage: hadoop_import.pl <collection> [-debug] [-dry_run] [-start_thrift] [-disable_thrift] [-refresh_import] [-flush_diskcache] [-use_nfs]\n\n";
+  print STDERR "usage: hadoop_import.pl <collection> [-debug] [-disable_thrift] [-dry_run] [-start_thrift] [-refresh_import] [-flush_diskcache] [-use_nfs] [-stagger]\n";
+  print STDERR "where: [debug] print more debug messages to STDERR\n";
+  print STDERR "       [dry_run] don't actually perform an file actions\n";
   exit;
 }
@@ -65 +68 @@
   {
     $refresh_import = 1;
+  }
+  if ($ARGV[$offset] eq '-stagger')
+  {
+    $stagger = 1;
   }
   if ($ARGV[$offset] eq '-flush_diskcache')
@@ -255 +262 @@
 }
 
+# - If we've been asked to Stagger start-up, add "delay.me" files to the
+#   compute nodes
+if ($is_rocks_cluster && $stagger)
+{
+  &shellCommand('rocks run host "touch /tmp/greenstone/delay.me && chmod a+rw /tmp/greenstone/delay.me"');
+}
+
 # 3. Special case for *Server type infodbs (namely TDBServer and GDBMServer)
 #    where we start the server now to ensure it lives on the head node
@@ -295 +309 @@
 
 # 3.5 Start up the thrift server(s) if we've been asked to
+my $thrift_log = $gs_results_dir . '/thriftctl.log';
 if ($start_thrift)
 {
@@ -300 +315 @@
   {
     print " * Starting Thrift Servers (on compute nodes)... ";
-    print "[DEBUG]\n" . &shellCommand('rocks run host "cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && ./thriftctl.sh start"') . "\n\n";
+    &shellCommand('rocks run host "cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && ./thriftctl.sh start" > "' . $thrift_log . '" 2>&1');
   }
   # single server
@@ -306 +321 @@
   {
     print " * Starting Thrift Server... ";
-    &shellCommand('cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && thriftctl.sh start');
+    &shellCommand('cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && thriftctl.sh start > "' . $thrift_log . '" 2>&1');
   }
   print "Done!\n";
@@ -362 +377 @@
   {
     print " * Stopping Thrift Servers (on compute nodes)... ";
-    &shellCommand('rocks run host "cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && ./thriftctl.sh stop"');
+    &shellCommand('rocks run host "cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && ./thriftctl.sh stop" >> "' . $thrift_log . '" 2>&1 ');
   }
   # single server
@@ -368 +383 @@
   {
     print " * Stoping Thrift Server... ";
-    &shellCommand('cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && thriftctl.sh start');
+    &shellCommand('cd ' . $ENV{'GEXTPARALLELBUILDING'} . '/packages/ThriftFS-0.9.0/bin && thriftctl.sh start >> "' . $thrift_log . '" 2>&1');
   }
   print "Done!\n";
@@ -384 +399 @@
   &shellCommand('cp ' . $gs_collection_dir . '/logs/*.* ' . $gs_results_dir);
 }
+if ($start_thrift && -d '/tmp/thrift')
+{
+  &shellCommand('cp /tmp/thrift/*.log ' . $gs_results_dir);
+}
 # - remote files
 if ($is_rocks_cluster)
 {
   &shellCommand('rocks run host "scp /tmp/greenstone/*.* ' . $cluster_head . ':' . $gs_results_dir . '"');
-&shellCommand('rocks run host "scp /tmp/gsimport-*/logs/*.log ' . $cluster_head . ':' . $gs_results_dir . '"');
+  &shellCommand('rocks run host "scp /tmp/gsimport-*/logs/*.log ' . $cluster_head . ':' . $gs_results_dir . '"');
+  if ($start_thrift)
+  {
+    &shellCommand('rocks run host "scp /tmp/thrift/*.log ' . $cluster_head . ':' . $gs_results_dir . '"');
+  }
 }
 print "Done!\n";
-# - generate data locality report
+
+# - generate data locality report...
 &shellCommand('parse_task_info_from_hadoop_log.pl "' . $gs_results_dir . '"');
+
+# - hadoop report...
+&shellCommand('hadoop_report.pl "' . $gs_results_dir . '"');
+
+# - and gantt chart
+&shellCommand('generate_gantt.pl -width 1600 "' . $gs_results_dir . '"');
 
 # 7. Done - clean up

gs2-extensions/parallel-building/trunk/src/src/java/org/nzdl/gsdl/HadoopGreenstoneIngest.java

@@ -16 +16 @@
 import java.net.InetAddress;
 import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.hadoop.fs.Path;
@@ -190 +192 @@
       // - open a unique log file
       File import_process_log = new File("/tmp/greenstone/import-hadoop-" + task_id + ".log");
+      FileWriter fw1 = new FileWriter(import_process_log, true);
+      // MEDUSA Customization: Introduce a slight delay based upon the hostname
+      // in order to stagger the startup of Map workers. It looks like the avg
+      // IO is around 25 minutes... so lets try to make it so the last mapper
+      // starts up 25 minutes after the first (with all others spread in
+      // between).
+      String hostname = InetAddress.getLocalHost().getHostName();
+      // We only do this if there is a sentinel file lurking in tmp
+      try
+      {
+        File delay_file = new File("/tmp/greenstone/delay.me");
+        if (delay_file.exists())
+        {
+          Pattern p = Pattern.compile("compute-0-([0-9]+).local");
+          Matcher m = p.matcher(hostname);
+          if (m.matches())
+          {
+            String node_str = m.group(1);
+            int node_number = Integer.parseInt(node_str) * 100;
+            fw1.write("[DEBUG] Delaying start for " + node_number + " seconds");
+            Thread.currentThread().sleep(1000 * node_number);
+          }
+          // We only do this once for each compute node
+          delay_file.delete();
+        }
+      }
+      catch (Exception ie)
+      {
+        System.err.println(ie.toString());
+      }
+
       // - start the log by writing the time and the manifest line
-      FileWriter fw1 = new FileWriter(import_process_log, true);
       long start_time = System.currentTimeMillis()/1000;
       StringBuffer header_block = new StringBuffer("[Started:");
       header_block.append(start_time);
       header_block.append("]\n[Host:");
-      header_block.append(InetAddress.getLocalHost().getHostName());
+      header_block.append(hostname);
       header_block.append("]\n[CPU:");
       String getcpu_executable_cmd = gsdlhome + "/ext/parallel-building/linux/bin/getcpu";