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

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

File:

• gs2-extensions/parallel-building/trunk/src/src/java/org/nzdl/gsdl/HadoopGreenstoneIngest.java (modified) (2 diffs)

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";