Codewut Because Software Matters

#replace eth1:0 with your NIC / alias
bind_ip=$(/sbin/ifconfig eth1:0 | grep 'inet addr:' | cut -d: -f2 | awk '{print $1}')
 
export BIND_OPTS="-Dlocal.bind.address=${bind_ip}"
 
# Command specific options appended to HADOOP_OPTS when specified
export HADOOP_NAMENODE_OPTS="-Dcom.sun.management.jmxremote $HADOOP_NAMENODE_OPTS $BIND_OPTS"
export HADOOP_SECONDARYNAMENODE_OPTS="-Dcom.sun.management.jmxremote $HADOOP_SECONDARYNAMENODE_OPTS $BIND_OPTS"
export HADOOP_DATANODE_OPTS="-Dcom.sun.management.jmxremote $HADOOP_DATANODE_OPTS $BIND_OPTS"
export HADOOP_BALANCER_OPTS="-Dcom.sun.management.jmxremote $HADOOP_BALANCER_OPTS $BIND_OPTS" 
export HADOOP_JOBTRACKER_OPTS="-Dcom.sun.management.jmxremote $HADOOP_JOBTRACKER_OPTS $BIND_OPTS"

<property>
  <name>dfs.secondary.http.address</name>
  <value>${local.bind.address}:50090</value>
  <description>
    The secondary namenode http server address and port.
    If the port is 0 then the server will start on a free port.
  </description>
</property>
 
<property>
  <name>dfs.datanode.address</name>
  <value>${local.bind.address}:50010</value>
</property>
 
<property>
  <name>dfs.datanode.http.address</name>
  <value>${local.bind.address}:50075</value>
</property>
 
<property>
  <name>dfs.datanode.ipc.address</name>
  <value>${local.bind.address}:50020</value>
</property>
 
<property>
  <name>dfs.http.address</name>
  <value>${local.bind.address}:50070</value>
</property>
 
<property>
  <name>dfs.datanode.https.address</name>
  <value>${local.bind.address}:50475</value>
</property>
 
<property>
  <name>dfs.https.address</name>
  <value>${local.bind.address}:50470</value>
</property>

public class TestThreadPool extends TestCase
{
 private static Log log = LogFactory.getLog(TestThreadPool.class);
 
 public void testLoadTracker() throws InterruptedException
 {
  int maxRunningThreads = 128;
  int maxLoad = 500;
 
  LoadTracker load = new LoadTracker(maxLoad);
  ExecutorService pool = Executors.newFixedThreadPool(maxRunningThreads);
 
  for (int i = 0; i < 500; i++)
  {
   // here you would create your real job and *predict* its impact on the load factor.
   // we choose the load to be random.
   int predictedJobLoad = (int) Math.round(Math.random() * 10l);
   MyJob aJob = new MyJob(load, predictedJobLoad,"job-"+i,this);
 
   while (!load.canHandle(predictedJobLoad))
   {
    log.debug(String.format("WAIT: current load %d and new job is about to be %d", load.get(), predictedJobLoad));
    synchronized (this) { this.wait(1000); }
   }
 
   log.debug(String.format("QUEUE: current load is %d and new job is about to be %d", load.get(), predictedJobLoad));
   pool.execute(aJob);
  }
  pool.shutdown();
  pool.awaitTermination(42,TimeUnit.DAYS);
  assertEquals(0, load.get());
 }
 
 private  class MyJob implements Runnable
 {
  private LoadTracker loadTracker;
  private int load;
  private String jobId;
  private Object monitor;
 
  public MyJob(LoadTracker loadTracker, int load, String jobId, Object monitor)
  {
   this.jobId = jobId;
   this.loadTracker = loadTracker;
   this.load = load;
   this.monitor = monitor;
   loadTracker.add(load);
  }
 
  @Override
  public void run()
  {
   log.debug(String.format("RUN: %s with a load of %d", jobId, load));
   try
   {
    Thread.sleep((int) Math.round(Math.random() * 5000l));
   }
   catch (InterruptedException e)
   {
    e.printStackTrace();
   }
   log.debug(String.format("FIN: %s with a load of %d", jobId, load));
   loadTracker.remove(load);
   if (monitor != null) synchronized (monitor) { monitor.notify(); }
  }
 }
 
 private class LoadTracker
 {
  private int currentLoad = 0;
  private int maxLoad = 0;
 
  public LoadTracker(int maxLoad)
  {
   this.maxLoad = maxLoad;
  }
 
  private synchronized void add(int load)
  {
   this.currentLoad += load;
  }
 
  private synchronized void remove(int load)
  {
   this.currentLoad -= load;
  }
 
  public synchronized int get()
  {
   return currentLoad;
  }
 
  public synchronized boolean canHandle(int additionalLoad)
  {
   return ((this.get() + additionalLoad) < maxLoad);
  }
 }
}

me@host:/mnt/data1/import$ cat bigfile.txt | pv | pigz -c | ssh me@otherhost 'cat | unpigz > /mnt/data1/bigfile.txt'
1.83GB 0:00:18 [95.3MB/s] [                            <=>                                                   ]