<commit_before>package water;

import java.io.*;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

import jsr166y.*;
import water.exec.Function;
import water.hdfs.HdfsLoader;
import water.nbhm.NonBlockingHashMap;
import water.store.s3.PersistS3;
import water.util.Utils;

import com.amazonaws.auth.PropertiesCredentials;
import com.google.common.base.Objects;
import com.google.common.io.Closeables;

/**
* Start point for creating or joining an <code>H2O</code> Cloud.
*
* @author <a href="mailto:<EMAIL>"></a>
* @version 1.0
*/
public final class H2O {

  static boolean _hdfsActive = false;

  static final String VERSION = "v0.3";

  // User name for this Cloud
  public static String NAME;

  // The default port for finding a Cloud
  static final int DEFAULT_PORT = 54321;
  public static int UDP_PORT; // Fast/small UDP transfers
  public static int API_PORT; // RequestServer and the new API HTTP port

  // The multicast discovery port
  static MulticastSocket  CLOUD_MULTICAST_SOCKET;
  static NetworkInterface CLOUD_MULTICAST_IF;
  static InetAddress      CLOUD_MULTICAST_GROUP;
  static int              CLOUD_MULTICAST_PORT ;
  // Default NIO Datagram channel
  static DatagramChannel  CLOUD_DGRAM;

  // Myself, as a Node in the Cloud
  public static H2ONode SELF = null;

  // Initial arguments
  public static String[] ARGS;

  public static final PrintStream OUT = System.out;
  public static final PrintStream ERR = System.err;
  public static final int NUMCPUS = Runtime.getRuntime().availableProcessors();

  // Convenience error
  public static final RuntimeException unimpl() { return new RuntimeException("unimplemented"); }

  // Central /dev/null for ignored exceptions
  public static final void ignore(Throwable e)             { ignore(e,"[h2o] Problem ignored: "); }
  public static final void ignore(Throwable e, String msg) { ignore(e, msg, true); }
  public static final void ignore(Throwable e, String msg, boolean printException) {
    StringBuffer sb = new StringBuffer();
    sb.append(msg).append('\n');
    if (printException) {
      StackTraceElement[] stack = e.getStackTrace();
      // The replacement of Exception -> Problem is required by our testing framework which would report
      // error if it sees "exception" in the node output
      sb.append(e.toString().replace("Exception", "Problem")).append('\n');
      for (StackTraceElement el : stack) { sb.append("\tat "); sb.append(el.toString().replace("Exception", "Problem" )); sb.append('\n'); }
    }
    System.err.println(sb);
  }

  // --------------------------------------------------------------------------
  // The Current Cloud. A list of all the Nodes in the Cloud. Changes if we
  // decide to change Clouds via atomic Cloud update.
  static public volatile H2O CLOUD = new H2O(new H2ONode[0],0,0);

  // ---
  // A dense array indexing all Cloud members. Fast reversal from "member#" to
  // Node.  No holes.  Cloud size is _members.length.
  public final H2ONode[] _memary;
  public final int _hash;

  // A dense integer identifier that rolls over rarely. Rollover limits the
  // number of simultaneous nested Clouds we are operating on in-parallel.
  // Really capped to 1 byte, under the assumption we won't have 256 nested
  // Clouds. Capped at 1 byte so it can be part of an atomically-assigned
  // 'long' holding info specific to this Cloud.
  public final char _idx; // no unsigned byte, so unsigned char instead

  // Is nnn larger than old (counting for wrap around)? Gets confused if we
  // start seeing a mix of more than 128 unique clouds at the same time. Used
  // to tell the order of Clouds appearing.
  static public boolean larger( int nnn, int old ) {
    assert (0 <= nnn && nnn <= 255);
    assert (0 <= old && old <= 255);
    return ((nnn-old)&0xFF) < 64;
  }

  // Static list of acceptable Cloud members
  static HashSet<H2ONode> STATIC_H2OS = null;

  // Reverse cloud index to a cloud; limit of 256 old clouds.
  static private final H2O[] CLOUDS = new H2O[256];

  // Construct a new H2O Cloud from the member list
  public H2O( H2ONode[] h2os, int hash, int idx ) {
    _memary = h2os;             // Need to clone?
    Arrays.sort(_memary);       // ... sorted!
    _hash = hash;               // And record hash for cloud rollover
    _idx = (char)(idx&0x0ff);   // Roll-over at 256
  }

  // One-shot atomic setting of the next Cloud, with an empty K/V store.
  // Called single-threaded from Paxos. Constructs the new H2O Cloud from a
  // member list.
  void set_next_Cloud( H2ONode[] h2os, int hash ) {
    synchronized(this) {
      int idx = _idx+1; // Unique 1-byte Cloud index
      if( idx == 256 ) idx=1; // wrap, avoiding zero
      CLOUDS[idx] = CLOUD = new H2O(h2os,hash,idx);
    }
    Paxos.print("Announcing new Cloud Membership: ",_memary);
  }

  // Check if the cloud id matches with one of the old clouds
  static boolean isIDFromPrevCloud(H2ONode h2o) {
    if ( h2o == null ) return false;
    HeartBeat hb = h2o._heartbeat;
    int hash = hb._cloud_hash;
    for( int i=0; i < 256; i++ )
      if( CLOUDS[i] != null && hash == CLOUDS[i]._hash )
        return true;
    return false;
  }

  public final int size() { return _memary.length; }
  public final H2ONode leader() { return _memary[0]; }

  // *Desired* distribution function on keys & replication factor. Replica #0
  // is the master, replica #1, 2, 3, etc represent additional desired
  // replication nodes. Note that this function is just the distribution
  // function - it does not DO any replication, nor does it dictate any policy
  // on how fast replication occurs. Returns -1 if the desired replica
  // is nonsense, e.g. asking for replica #3 in a 2-Node system.
  public int D( Key key, int repl ) {
    if( repl >= size() ) return -1;

    // See if this is a specifically homed Key
    byte[] kb = key._kb;
    if( !key.user_allowed() && repl < kb[1] ) { // Asking for a replica# from the homed list?
      H2ONode h2o=null, h2otmp = new H2ONode(); // Fill in the fields from the Key
      AutoBuffer ab = new AutoBuffer(kb,2);
      for( int i=0; i<=repl; i++ )
        h2o = h2otmp.read(ab);  // Read util we get the specified H2O
      // Reverse the home to the index
      int idx = nidx(h2o);
      if( idx >= 0 ) return idx;
      // Else homed to a node which is no longer in the cloud!
      // Fall back to the normal home mode
    }

    // Easy Cheesy Stupid:
    return ((key._hash+repl)&0x7FFFFFFF) % size();
  }

  // Find the node index for this H2ONode, or a negative number on a miss
  public int nidx( H2ONode h2o ) { return Arrays.binarySearch(_memary,h2o); }
  public boolean contains( H2ONode h2o ) { return nidx(h2o) >= 0; }
  @Override public String toString() {
    return Arrays.toString(_memary);
  }

  static InetAddress findInetAddressForSelf() throws Error {
    // Get a list of all valid IPs on this machine.  Typically 1 on Mac or
    // Windows, but could be many on Linux or if a hypervisor is present.
    ArrayList<InetAddress> ips = new ArrayList<InetAddress>();
    try {
      Enumeration<NetworkInterface> nis = NetworkInterface.getNetworkInterfaces();
      while( nis.hasMoreElements() ) {
        NetworkInterface ni = nis.nextElement();
        Enumeration<InetAddress> ias = ni.getInetAddresses();
        while( ias.hasMoreElements() ) {
          ips.add(ias.nextElement());
        }
      }
    } catch( SocketException e ) { }

    InetAddress local = null;   // My final choice

    // Check for an "-ip xxxx" option and accept a valid user choice; required
    // if there are multiple valid IP addresses.
    InetAddress arg = null;
    if (OPT_ARGS.ip != null) {
      try{
        arg = InetAddress.getByName(OPT_ARGS.ip);
      } catch( UnknownHostException e ) {
        System.err.println(e.toString());
        System.exit(-1);
      }
      if( !(arg instanceof Inet4Address) ) {
        System.err.println("Only IP4 addresses allowed.");
        System.exit(-1);
      }
      if( !ips.contains(arg) ) {
        System.err.println("IP address not found on this machine");
        System.exit(-1);
      }
      local = arg;
    } else {
      // No user-specified IP address.  Attempt auto-discovery.  Roll through
      // all the network choices on looking for a single Inet4.
      ArrayList<InetAddress> validIps = new ArrayList();
      for( InetAddress ip : ips ) {
        // make sure the given IP address can be found here
        if( ip instanceof Inet4Address &&
            !ip.isLoopbackAddress() &&
            !ip.isLinkLocalAddress() ) {
          validIps.add(ip);
        }
      }
      if( validIps.size() == 1 ) {
        local = validIps.get(0);
      } else {
        local = guessInetAddress(validIps);
      }
    }

    // The above fails with no network connection, in that case go for a truly
    // local host.
    if( local == null ) {
      try {
        System.err.println("Failed to determine IP, falling back to localhost.");
        // set default ip address to be 127.0.0.1 /localhost
        local = InetAddress.getByName("127.0.0.1");
      } catch( UnknownHostException e ) {
        throw new Error(e);
      }
    }
    return local;
  }

  private static InetAddress guessInetAddress(List<InetAddress> ips) {
    System.out.println("Multiple local IPs detected:");
    for(InetAddress ip : ips) System.out.println("  " + ip);
    System.out.println("Attempting to determine correct address...");
    Socket s = null;
    try {
      // using google's DNS server as an external IP to find
      s = new Socket("8.8.8.8", 53);
      System.out.println("Using " + s.getLocalAddress());
      return s.getLocalAddress();
    } catch( Throwable t ) {
      return null;
    } finally {
      Utils.close(s);
    }
  }

  // --------------------------------------------------------------------------
  // The (local) set of Key/Value mappings.
  static final NonBlockingHashMap<Key,Value> STORE = new NonBlockingHashMap<Key, Value>();

  // Dummy shared volatile for ordering games
  static public volatile int VOLATILE;

  // PutIfMatch
  // - Atomically update the STORE, returning the old Value on success
  // - Kick the persistence engine as needed
  // - Return existing Value on fail, no change.
  //
  // Keys are interned here: I always keep the existing Key, if any. The
  // existing Key is blind jammed into the Value prior to atomically inserting
  // it into the STORE and interning.
  //
  // Because of the blind jam, there is a narrow unusual race where the Key
  // might exist but be stale (deleted, mapped to a TOMBSTONE), a fresh put()
  // can find it and jam it into the Value, then the Key can be deleted
  // completely (e.g. via an invalidate), the table can resize flushing the
  // stale Key, an unrelated weak-put can re-insert a matching Key (but as a
  // new Java object), and delete it, and then the original thread can do a
  // successful put_if_later over the missing Key and blow the invariant that a
  // stored Value always points to the physically equal Key that maps to it
  // from the STORE. If this happens, some of replication management bits in
  // the Key will be set in the wrong Key copy... leading to extra rounds of
  // replication.

  public static final Value putIfMatch( Key key, Value val, Value old ) {
    assert val==null || val._key == key; // Keys matched
    if( old != null && val != null ) // Have an old value?
      key = val._key = old._key; // Use prior key in val

    // Insert into the K/V store
    Value res = STORE.putIfMatchUnlocked(key,val,old);
    assert chk_equals_key(res, old);
    if( res != old ) return res; // Return the failure cause
    assert chk_equals_key(res, old, val);
    // Persistence-tickle.
    // If the K/V mapping is going away, remove the old guy.
    // If the K/V mapping is changing, let the store cleaner just overwrite.
    // If the K/V mapping is new, let the store cleaner just create
    if( old != null && val == null ) old.removeIce(); // Remove the old guy
    if( val != null ) dirty_store(); // Start storing the new guy
    return old; // Return success
  }

  // assert that all of val, old & res that are not-null all agree on key.
  private static final boolean chk_equals_key( Value... vs ) {
    Key k = null;
    for( Value v : vs ) {
      if( v != null ) {
        assert k == null || k == v._key;
        k = v._key;
      }
    }
    return true;
  }

  // Raw put; no marking the memory as out-of-sync with disk. Used to import
  // initial keys from local storage, or to intern keys.
  public static final Value putIfAbsent_raw( Key key, Value val ) {
    assert val.isSameKey(key);
    Value res = STORE.putIfMatchUnlocked(key,val,null);
    assert res == null;
    return res;
  }

  // Get the value from the store
  public static Value get( Key key ) {
    Value v = STORE.get(key);
    // Lazily manifest array chunks, if the backing file exists.
    if( v == null ) {
      v = Value.lazyArrayChunk(key);
      if( v == null ) return null;
      // Insert the manifested value, as-if it existed all along
      Value res = putIfMatch(key,v,null);
      if( res != null ) v = res; // This happens racily, so take any prior result
    }
    if( v != null ) v.touch();
    return v;
  }

  public static Value raw_get( Key key ) { return STORE.get(key); }
  public static Key getk( Key key ) { return STORE.getk(key); }
  public static Set<Key> keySet( ) { return STORE.keySet(); }
  public static Collection<Value> values( ) { return STORE.values(); }
  public static int store_size() { return STORE.size(); }


  // --------------------------------------------------------------------------
  // The worker pools - F/J pools with different priorities.

  // These priorities are carefully ordered and asserted for... modify with
  // care.  The real problem here is that we can get into cyclic deadlock
  // unless we spawn a thread of priority "X+1" in order to allow progress
  // on a queue which might be flooded with a large number of "<=X" tasks.
  //
  // Example of deadlock: suppose TaskPutKey and the Invalidate ran at the same
  // priority on a 2-node cluster.  Both nodes flood their own queues with
  // writes to unique keys, which require invalidates to run on the other node.
  // Suppose the flooding depth exceeds the thread-limit (e.g. 99); then each
  // node might have all 99 worker threads blocked in TaskPutKey, awaiting
  // remote invalidates - but the other nodes' threads are also all blocked
  // awaiting invalidates!
  //
  // We fix this by being willing to always spawn a thread working on jobs at
  // priority X+1, and guaranteeing there are no jobs above MAX_PRIORITY -
  // i.e., jobs running at MAX_PRIORITY cannot block, and when those jobs are
  // done, the next lower level jobs get unblocked, etc.
  public static final byte        MAX_PRIORITY = Byte.MAX_VALUE-1;
  public static final byte    ACK_ACK_PRIORITY = MAX_PRIORITY-0;
  public static final byte        ACK_PRIORITY = MAX_PRIORITY-1;
  public static final byte   DESERIAL_PRIORITY = MAX_PRIORITY-2;
  public static final byte INVALIDATE_PRIORITY = MAX_PRIORITY-2;
  public static final byte    ARY_KEY_PRIORITY = MAX_PRIORITY-2;
  public static final byte    GET_KEY_PRIORITY = MAX_PRIORITY-3;
  public static final byte    PUT_KEY_PRIORITY = MAX_PRIORITY-4;
  public static final byte     ATOMIC_PRIORITY = MAX_PRIORITY-5;
  public static final byte        GUI_PRIORITY = MAX_PRIORITY-6;
  public static final byte     MIN_HI_PRIORITY = MAX_PRIORITY-6;
  public static final byte        MIN_PRIORITY = 0;

  // F/J threads that remember the priority of the last task they started
  // working on.
  static class FJWThr extends ForkJoinWorkerThread {
    public int _priority;
    FJWThr(ForkJoinPool pool) { super(pool); }
  }
  // Factory for F/J threads, with cap's that vary with priority.
  static class FJWThrFact implements ForkJoinPool.ForkJoinWorkerThreadFactory {
    private final int _cap;
    FJWThrFact( int cap ) { _cap = cap; }
    public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
      return pool.getPoolSize() <= _cap ? new FJWThr(pool) : null;
    }
  }

  // A standard FJ Pool, with an expected priority level.
  private static class ForkJoinPool2 extends ForkJoinPool {
    public final int _priority;
    ForkJoinPool2(int p, int cap) { super(NUMCPUS,new FJWThrFact(cap),null,p!=MIN_PRIORITY); _priority = p; }
    public H2OCountedCompleter poll() { return (H2OCountedCompleter)pollSubmission(); }
  }

  // Normal-priority work is generally directly-requested user ops.
  private static final ForkJoinPool2 FJP_NORM = new ForkJoinPool2(MIN_PRIORITY,99);
  // Hi-priority work, sorted into individual queues per-priority.
  // Capped at a small number of threads per pool.
  private static final ForkJoinPool2 FJPS[] = new ForkJoinPool2[MAX_PRIORITY+1];
  static {
    // Only need 1 thread for the AckAck work, as it cannot block
    FJPS[ACK_ACK_PRIORITY] = new ForkJoinPool2(ACK_ACK_PRIORITY,1);
    for( int i=MIN_HI_PRIORITY+1; i<MAX_PRIORITY; i++ )
      FJPS[i] = new ForkJoinPool2(i,NUMCPUS); // All CPUs, but no more for blocking purposes
    FJPS[GUI_PRIORITY] = new ForkJoinPool2(GUI_PRIORITY,2);
    FJPS[0] = FJP_NORM;
  }

  // Easy peeks at the low FJ queue
  public static int getLoQueue (     ) { return FJP_NORM.getQueuedSubmissionCount();}
  public static int loQPoolSize(     ) { return FJP_NORM.getPoolSize();             }
  public static int getHiQueue (int i) { return FJPS[i+MIN_HI_PRIORITY].getQueuedSubmissionCount();}
  public static int hiQPoolSize(int i) { return FJPS[i+MIN_HI_PRIORITY].getPoolSize();             }

  // Submit to the correct priority queue
  public static void submitTask( H2OCountedCompleter task ) {
    int priority = task.priority();<commit_msg>Raise default FJ pool size to 299<commit_after>package water;

import java.io.*;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

import jsr166y.*;
import water.exec.Function;
import water.hdfs.HdfsLoader;
import water.nbhm.NonBlockingHashMap;
import water.store.s3.PersistS3;
import water.util.Utils;

import com.amazonaws.auth.PropertiesCredentials;
import com.google.common.base.Objects;
import com.google.common.io.Closeables;

/**
* Start point for creating or joining an <code>H2O</code> Cloud.
*
* @author <a href="mailto:<EMAIL>"></a>
* @version 1.0
*/
public final class H2O {

  static boolean _hdfsActive = false;

  static final String VERSION = "v0.3";

  // User name for this Cloud
  public static String NAME;

  // The default port for finding a Cloud
  static final int DEFAULT_PORT = 54321;
  public static int UDP_PORT; // Fast/small UDP transfers
  public static int API_PORT; // RequestServer and the new API HTTP port

  // The multicast discovery port
  static MulticastSocket  CLOUD_MULTICAST_SOCKET;
  static NetworkInterface CLOUD_MULTICAST_IF;
  static InetAddress      CLOUD_MULTICAST_GROUP;
  static int              CLOUD_MULTICAST_PORT ;
  // Default NIO Datagram channel
  static DatagramChannel  CLOUD_DGRAM;

  // Myself, as a Node in the Cloud
  public static H2ONode SELF = null;

  // Initial arguments
  public static String[] ARGS;

  public static final PrintStream OUT = System.out;
  public static final PrintStream ERR = System.err;
  public static final int NUMCPUS = Runtime.getRuntime().availableProcessors();

  // Convenience error
  public static final RuntimeException unimpl() { return new RuntimeException("unimplemented"); }

  // Central /dev/null for ignored exceptions
  public static final void ignore(Throwable e)             { ignore(e,"[h2o] Problem ignored: "); }
  public static final void ignore(Throwable e, String msg) { ignore(e, msg, true); }
  public static final void ignore(Throwable e, String msg, boolean printException) {
    StringBuffer sb = new StringBuffer();
    sb.append(msg).append('\n');
    if (printException) {
      StackTraceElement[] stack = e.getStackTrace();
      // The replacement of Exception -> Problem is required by our testing framework which would report
      // error if it sees "exception" in the node output
      sb.append(e.toString().replace("Exception", "Problem")).append('\n');
      for (StackTraceElement el : stack) { sb.append("\tat "); sb.append(el.toString().replace("Exception", "Problem" )); sb.append('\n'); }
    }
    System.err.println(sb);
  }

  // --------------------------------------------------------------------------
  // The Current Cloud. A list of all the Nodes in the Cloud. Changes if we
  // decide to change Clouds via atomic Cloud update.
  static public volatile H2O CLOUD = new H2O(new H2ONode[0],0,0);

  // ---
  // A dense array indexing all Cloud members. Fast reversal from "member#" to
  // Node.  No holes.  Cloud size is _members.length.
  public final H2ONode[] _memary;
  public final int _hash;

  // A dense integer identifier that rolls over rarely. Rollover limits the
  // number of simultaneous nested Clouds we are operating on in-parallel.
  // Really capped to 1 byte, under the assumption we won't have 256 nested
  // Clouds. Capped at 1 byte so it can be part of an atomically-assigned
  // 'long' holding info specific to this Cloud.
  public final char _idx; // no unsigned byte, so unsigned char instead

  // Is nnn larger than old (counting for wrap around)? Gets confused if we
  // start seeing a mix of more than 128 unique clouds at the same time. Used
  // to tell the order of Clouds appearing.
  static public boolean larger( int nnn, int old ) {
    assert (0 <= nnn && nnn <= 255);
    assert (0 <= old && old <= 255);
    return ((nnn-old)&0xFF) < 64;
  }

  // Static list of acceptable Cloud members
  static HashSet<H2ONode> STATIC_H2OS = null;

  // Reverse cloud index to a cloud; limit of 256 old clouds.
  static private final H2O[] CLOUDS = new H2O[256];

  // Construct a new H2O Cloud from the member list
  public H2O( H2ONode[] h2os, int hash, int idx ) {
    _memary = h2os;             // Need to clone?
    Arrays.sort(_memary);       // ... sorted!
    _hash = hash;               // And record hash for cloud rollover
    _idx = (char)(idx&0x0ff);   // Roll-over at 256
  }

  // One-shot atomic setting of the next Cloud, with an empty K/V store.
  // Called single-threaded from Paxos. Constructs the new H2O Cloud from a
  // member list.
  void set_next_Cloud( H2ONode[] h2os, int hash ) {
    synchronized(this) {
      int idx = _idx+1; // Unique 1-byte Cloud index
      if( idx == 256 ) idx=1; // wrap, avoiding zero
      CLOUDS[idx] = CLOUD = new H2O(h2os,hash,idx);
    }
    Paxos.print("Announcing new Cloud Membership: ",_memary);
  }

  // Check if the cloud id matches with one of the old clouds
  static boolean isIDFromPrevCloud(H2ONode h2o) {
    if ( h2o == null ) return false;
    HeartBeat hb = h2o._heartbeat;
    int hash = hb._cloud_hash;
    for( int i=0; i < 256; i++ )
      if( CLOUDS[i] != null && hash == CLOUDS[i]._hash )
        return true;
    return false;
  }

  public final int size() { return _memary.length; }
  public final H2ONode leader() { return _memary[0]; }

  // *Desired* distribution function on keys & replication factor. Replica #0
  // is the master, replica #1, 2, 3, etc represent additional desired
  // replication nodes. Note that this function is just the distribution
  // function - it does not DO any replication, nor does it dictate any policy
  // on how fast replication occurs. Returns -1 if the desired replica
  // is nonsense, e.g. asking for replica #3 in a 2-Node system.
  public int D( Key key, int repl ) {
    if( repl >= size() ) return -1;

    // See if this is a specifically homed Key
    byte[] kb = key._kb;
    if( !key.user_allowed() && repl < kb[1] ) { // Asking for a replica# from the homed list?
      H2ONode h2o=null, h2otmp = new H2ONode(); // Fill in the fields from the Key
      AutoBuffer ab = new AutoBuffer(kb,2);
      for( int i=0; i<=repl; i++ )
        h2o = h2otmp.read(ab);  // Read util we get the specified H2O
      // Reverse the home to the index
      int idx = nidx(h2o);
      if( idx >= 0 ) return idx;
      // Else homed to a node which is no longer in the cloud!
      // Fall back to the normal home mode
    }

    // Easy Cheesy Stupid:
    return ((key._hash+repl)&0x7FFFFFFF) % size();
  }

  // Find the node index for this H2ONode, or a negative number on a miss
  public int nidx( H2ONode h2o ) { return Arrays.binarySearch(_memary,h2o); }
  public boolean contains( H2ONode h2o ) { return nidx(h2o) >= 0; }
  @Override public String toString() {
    return Arrays.toString(_memary);
  }

  static InetAddress findInetAddressForSelf() throws Error {
    // Get a list of all valid IPs on this machine.  Typically 1 on Mac or
    // Windows, but could be many on Linux or if a hypervisor is present.
    ArrayList<InetAddress> ips = new ArrayList<InetAddress>();
    try {
      Enumeration<NetworkInterface> nis = NetworkInterface.getNetworkInterfaces();
      while( nis.hasMoreElements() ) {
        NetworkInterface ni = nis.nextElement();
        Enumeration<InetAddress> ias = ni.getInetAddresses();
        while( ias.hasMoreElements() ) {
          ips.add(ias.nextElement());
        }
      }
    } catch( SocketException e ) { }

    InetAddress local = null;   // My final choice

    // Check for an "-ip xxxx" option and accept a valid user choice; required
    // if there are multiple valid IP addresses.
    InetAddress arg = null;
    if (OPT_ARGS.ip != null) {
      try{
        arg = InetAddress.getByName(OPT_ARGS.ip);
      } catch( UnknownHostException e ) {
        System.err.println(e.toString());
        System.exit(-1);
      }
      if( !(arg instanceof Inet4Address) ) {
        System.err.println("Only IP4 addresses allowed.");
        System.exit(-1);
      }
      if( !ips.contains(arg) ) {
        System.err.println("IP address not found on this machine");
        System.exit(-1);
      }
      local = arg;
    } else {
      // No user-specified IP address.  Attempt auto-discovery.  Roll through
      // all the network choices on looking for a single Inet4.
      ArrayList<InetAddress> validIps = new ArrayList();
      for( InetAddress ip : ips ) {
        // make sure the given IP address can be found here
        if( ip instanceof Inet4Address &&
            !ip.isLoopbackAddress() &&
            !ip.isLinkLocalAddress() ) {
          validIps.add(ip);
        }
      }
      if( validIps.size() == 1 ) {
        local = validIps.get(0);
      } else {
        local = guessInetAddress(validIps);
      }
    }

    // The above fails with no network connection, in that case go for a truly
    // local host.
    if( local == null ) {
      try {
        System.err.println("Failed to determine IP, falling back to localhost.");
        // set default ip address to be 127.0.0.1 /localhost
        local = InetAddress.getByName("127.0.0.1");
      } catch( UnknownHostException e ) {
        throw new Error(e);
      }
    }
    return local;
  }

  private static InetAddress guessInetAddress(List<InetAddress> ips) {
    System.out.println("Multiple local IPs detected:");
    for(InetAddress ip : ips) System.out.println("  " + ip);
    System.out.println("Attempting to determine correct address...");
    Socket s = null;
    try {
      // using google's DNS server as an external IP to find
      s = new Socket("8.8.8.8", 53);
      System.out.println("Using " + s.getLocalAddress());
      return s.getLocalAddress();
    } catch( Throwable t ) {
      return null;
    } finally {
      Utils.close(s);
    }
  }

  // --------------------------------------------------------------------------
  // The (local) set of Key/Value mappings.
  static final NonBlockingHashMap<Key,Value> STORE = new NonBlockingHashMap<Key, Value>();

  // Dummy shared volatile for ordering games
  static public volatile int VOLATILE;

  // PutIfMatch
  // - Atomically update the STORE, returning the old Value on success
  // - Kick the persistence engine as needed
  // - Return existing Value on fail, no change.
  //
  // Keys are interned here: I always keep the existing Key, if any. The
  // existing Key is blind jammed into the Value prior to atomically inserting
  // it into the STORE and interning.
  //
  // Because of the blind jam, there is a narrow unusual race where the Key
  // might exist but be stale (deleted, mapped to a TOMBSTONE), a fresh put()
  // can find it and jam it into the Value, then the Key can be deleted
  // completely (e.g. via an invalidate), the table can resize flushing the
  // stale Key, an unrelated weak-put can re-insert a matching Key (but as a
  // new Java object), and delete it, and then the original thread can do a
  // successful put_if_later over the missing Key and blow the invariant that a
  // stored Value always points to the physically equal Key that maps to it
  // from the STORE. If this happens, some of replication management bits in
  // the Key will be set in the wrong Key copy... leading to extra rounds of
  // replication.

  public static final Value putIfMatch( Key key, Value val, Value old ) {
    assert val==null || val._key == key; // Keys matched
    if( old != null && val != null ) // Have an old value?
      key = val._key = old._key; // Use prior key in val

    // Insert into the K/V store
    Value res = STORE.putIfMatchUnlocked(key,val,old);
    assert chk_equals_key(res, old);
    if( res != old ) return res; // Return the failure cause
    assert chk_equals_key(res, old, val);
    // Persistence-tickle.
    // If the K/V mapping is going away, remove the old guy.
    // If the K/V mapping is changing, let the store cleaner just overwrite.
    // If the K/V mapping is new, let the store cleaner just create
    if( old != null && val == null ) old.removeIce(); // Remove the old guy
    if( val != null ) dirty_store(); // Start storing the new guy
    return old; // Return success
  }

  // assert that all of val, old & res that are not-null all agree on key.
  private static final boolean chk_equals_key( Value... vs ) {
    Key k = null;
    for( Value v : vs ) {
      if( v != null ) {
        assert k == null || k == v._key;
        k = v._key;
      }
    }
    return true;
  }

  // Raw put; no marking the memory as out-of-sync with disk. Used to import
  // initial keys from local storage, or to intern keys.
  public static final Value putIfAbsent_raw( Key key, Value val ) {
    assert val.isSameKey(key);
    Value res = STORE.putIfMatchUnlocked(key,val,null);
    assert res == null;
    return res;
  }

  // Get the value from the store
  public static Value get( Key key ) {
    Value v = STORE.get(key);
    // Lazily manifest array chunks, if the backing file exists.
    if( v == null ) {
      v = Value.lazyArrayChunk(key);
      if( v == null ) return null;
      // Insert the manifested value, as-if it existed all along
      Value res = putIfMatch(key,v,null);
      if( res != null ) v = res; // This happens racily, so take any prior result
    }
    if( v != null ) v.touch();
    return v;
  }

  public static Value raw_get( Key key ) { return STORE.get(key); }
  public static Key getk( Key key ) { return STORE.getk(key); }
  public static Set<Key> keySet( ) { return STORE.keySet(); }
  public static Collection<Value> values( ) { return STORE.values(); }
  public static int store_size() { return STORE.size(); }


  // --------------------------------------------------------------------------
  // The worker pools - F/J pools with different priorities.

  // These priorities are carefully ordered and asserted for... modify with
  // care.  The real problem here is that we can get into cyclic deadlock
  // unless we spawn a thread of priority "X+1" in order to allow progress
  // on a queue which might be flooded with a large number of "<=X" tasks.
  //
  // Example of deadlock: suppose TaskPutKey and the Invalidate ran at the same
  // priority on a 2-node cluster.  Both nodes flood their own queues with
  // writes to unique keys, which require invalidates to run on the other node.
  // Suppose the flooding depth exceeds the thread-limit (e.g. 99); then each
  // node might have all 99 worker threads blocked in TaskPutKey, awaiting
  // remote invalidates - but the other nodes' threads are also all blocked
  // awaiting invalidates!
  //
  // We fix this by being willing to always spawn a thread working on jobs at
  // priority X+1, and guaranteeing there are no jobs above MAX_PRIORITY -
  // i.e., jobs running at MAX_PRIORITY cannot block, and when those jobs are
  // done, the next lower level jobs get unblocked, etc.
  public static final byte        MAX_PRIORITY = Byte.MAX_VALUE-1;
  public static final byte    ACK_ACK_PRIORITY = MAX_PRIORITY-0;
  public static final byte        ACK_PRIORITY = MAX_PRIORITY-1;
  public static final byte   DESERIAL_PRIORITY = MAX_PRIORITY-2;
  public static final byte INVALIDATE_PRIORITY = MAX_PRIORITY-2;
  public static final byte    ARY_KEY_PRIORITY = MAX_PRIORITY-2;
  public static final byte    GET_KEY_PRIORITY = MAX_PRIORITY-3;
  public static final byte    PUT_KEY_PRIORITY = MAX_PRIORITY-4;
  public static final byte     ATOMIC_PRIORITY = MAX_PRIORITY-5;
  public static final byte        GUI_PRIORITY = MAX_PRIORITY-6;
  public static final byte     MIN_HI_PRIORITY = MAX_PRIORITY-6;
  public static final byte        MIN_PRIORITY = 0;

  // F/J threads that remember the priority of the last task they started
  // working on.
  static class FJWThr extends ForkJoinWorkerThread {
    public int _priority;
    FJWThr(ForkJoinPool pool) { super(pool); }
  }
  // Factory for F/J threads, with cap's that vary with priority.
  static class FJWThrFact implements ForkJoinPool.ForkJoinWorkerThreadFactory {
    private final int _cap;
    FJWThrFact( int cap ) { _cap = cap; }
    public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
      return pool.getPoolSize() <= _cap ? new FJWThr(pool) : null;
    }
  }

  // A standard FJ Pool, with an expected priority level.
  private static class ForkJoinPool2 extends ForkJoinPool {
    public final int _priority;
    ForkJoinPool2(int p, int cap) { super(NUMCPUS,new FJWThrFact(cap),null,p!=MIN_PRIORITY); _priority = p; }
    public H2OCountedCompleter poll() { return (H2OCountedCompleter)pollSubmission(); }
  }

  // Normal-priority work is generally directly-requested user ops.
  private static final ForkJoinPool2 FJP_NORM = new ForkJoinPool2(MIN_PRIORITY,299);
  // Hi-priority work, sorted into individual queues per-priority.
  // Capped at a small number of threads per pool.
  private static final ForkJoinPool2 FJPS[] = new ForkJoinPool2[MAX_PRIORITY+1];
  static {
    // Only need 1 thread for the AckAck work, as it cannot block
    FJPS[ACK_ACK_PRIORITY] = new ForkJoinPool2(ACK_ACK_PRIORITY,1);
    for( int i=MIN_HI_PRIORITY+1; i<MAX_PRIORITY; i++ )
      FJPS[i] = new ForkJoinPool2(i,NUMCPUS); // All CPUs, but no more for blocking purposes
    FJPS[GUI_PRIORITY] = new ForkJoinPool2(GUI_PRIORITY,2);
    FJPS[0] = FJP_NORM;
  }

  // Easy peeks at the low FJ queue
  public static int getLoQueue (     ) { return FJP_NORM.getQueuedSubmissionCount();}
  public static int loQPoolSize(     ) { return FJP_NORM.getPoolSize();             }
  public static int getHiQueue (int i) { return FJPS[i+MIN_HI_PRIORITY].getQueuedSubmissionCount();}
  public static int hiQPoolSize(int i) { return FJPS[i+MIN_HI_PRIORITY].getPoolSize();             }

  // Submit to the correct priority queue
  public static void submitTask( H2OCountedCompleter task ) {
    int priority = task.priority();
{'fraction_non_alphanumeric': 0.1090584858870406, 'fraction_numerical': 0.010253838223411074, 'mean_word_length': 2.7210990150336962, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 6, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '748747', 'n_tokens_mistral': 11165, 'n_tokens_neox': 10744, 'n_words': 4081}