Added some print outs which will hopefully provide some useful info next time the mysterious Frame.split/rebalance exception hits.

tomasnykodym · tomasnykodym · commit da6b0e2b85a9 · 2014-09-09T15:01:38.000-07:00
diff --git a/src/main/java/water/fvec/CXIChunk.java b/src/main/java/water/fvec/CXIChunk.java
@@ -12,6 +12,7 @@
  * Sparse chunk.
  */
 public class CXIChunk extends Chunk {
+  protected transient int _sparseLen;            // Number of elements in this chunk
   protected transient int _valsz; // byte size of stored value
   protected transient int _valsz_log; //
   protected transient int _ridsz; // byte size of stored (chunk-relative) row nums
@@ -29,17 +30,18 @@ protected CXIChunk(int len, int nzs, int valsz, byte [] buf){
     assert valsz == 0 || (1 << log) == valsz;
     _valsz = valsz;
     _valsz_log = log;
-
     _ridsz = (len >= 65535)?4:2;
     UDP.set4(buf,0,len);
     byte b = (byte) _ridsz;
     buf[4] = b;
     buf[5] = (byte) _valsz;
     _mem = buf;
+    _sparseLen = (_mem.length - OFF) / (_valsz + _ridsz);
+    assert (_mem.length - OFF) % (_valsz + _ridsz) == 0:"unexpected mem.length in sparse chunk: mem.length = " + (_mem.length - OFF) + "val_sz = " + _valsz + ", rowId_sz = " + _ridsz;
   }
 
   @Override public final boolean isSparse() {return true;}
-  @Override public final int sparseLen(){return (_mem.length - OFF) / (_valsz + _ridsz);}
+  @Override public final int sparseLen(){return _sparseLen;}
   @Override public final int nonzeros(int [] arr){
     int len = sparseLen();
     int off = OFF;
@@ -77,6 +79,9 @@ protected CXIChunk(int len, int nzs, int valsz, byte [] buf){
 
   @Override boolean hasFloat ()                 { return false; }
 
+  @Override public String toString(){
+    return getClass().getSimpleName() + "( start = " + _start + ", len = " + _len + " sparseLen = " + _sparseLen + " valSz = " + _valsz + " rIdSz = " + _ridsz + ")";
+  }
   @Override NewChunk inflate_impl(NewChunk nc) {
     final int slen = sparseLen();
     nc.set_sparseLen(slen);
@@ -156,6 +161,8 @@ protected final int findOffset(int idx) {
     _len = UDP.get4(_mem,0);
     _ridsz = _mem[4];
     _valsz = _mem[5];
+    _sparseLen = (_mem.length - OFF) / (_valsz + _ridsz);
+    assert (_mem.length - OFF) % (_valsz + _ridsz) == 0:"unexpected mem.length in sparse chunk: mem.length = " + (_mem.length - OFF) + "val_sz = " + _valsz + ", rowId_sz = " + _ridsz;
     int x = _valsz;
     int log = 0;
     while(x > 1){
diff --git a/src/main/java/water/fvec/Chunk.java b/src/main/java/water/fvec/Chunk.java
@@ -244,7 +244,9 @@ protected int pformat_len0( double scale, int lg ) {
 
 
   @Override public Chunk clone() { return (Chunk)super.clone(); }
-  @Override public String toString() { return getClass().getSimpleName(); }
+  @Override public String toString() {
+    return getClass().getSimpleName() + "(start = " + _start + ", len = " + _len + ")";
+  }
 
   public long byteSize() {
     long s= _mem == null ? 0 : _mem.length;
diff --git a/src/main/java/water/fvec/ChunkSplitter.java b/src/main/java/water/fvec/ChunkSplitter.java
@@ -3,6 +3,7 @@
 import java.util.Iterator;
 
 import water.Futures;
+import water.util.Log;
 
 /** Helper to provide access to package
  * hidden methods and attributes.
@@ -16,32 +17,36 @@ public static NewChunk resetLen(NewChunk nc) {
 
   /** Extract portion of given chunk into given output chunk. */
   public static void extractChunkPart(Chunk ic, Chunk oc, int startRow, int nrows, Futures fs) {
-    NewChunk dst = new NewChunk(oc);
-    dst._len = dst._sparseLen = 0;
-    NewChunk src = new NewChunk(ic);
-    src = ic.inflate_impl(src);
-    assert src._len == ic._len;
-    // Iterate over values skip all 0
-    int remain = nrows;
-    Iterator<NewChunk.Value> it = src.values(startRow,startRow+nrows);
-    int off = startRow-1;
-    while(it.hasNext()) {
-      NewChunk.Value v = it.next();
-      final int rid = v.rowId0();
-      assert  rid < startRow+nrows;
-      int add = rid - off; // number of values to add
-      off = rid;
-      dst.addZeros(add-1); // append (add-1) zeros
-      v.add2Chunk(dst);    // followed by a value
-      remain -= add;
-      assert remain >= 0;
-    }
-    // Handle case when last added value is followed by zeros till startRow+nrows
-    dst.addZeros(remain);
-
-    assert dst._len == oc._len : "NewChunk.dst.len = " + dst._len + ", oc._len = " + oc._len;
-    dst.close(dst.cidx(),fs);
+    try {
+      NewChunk dst = new NewChunk(oc);
+      dst._len = dst._sparseLen = 0;
+      NewChunk src = new NewChunk(ic);
+      src = ic.inflate_impl(src);
+      assert src._len == ic._len;
+      // Iterate over values skip all 0
+      int remain = nrows;
+      Iterator<NewChunk.Value> it = src.values(startRow, startRow + nrows);
+      int off = startRow - 1;
+      while (it.hasNext()) {
+        NewChunk.Value v = it.next();
+        final int rid = v.rowId0();
+        assert rid < startRow + nrows;
+        int add = rid - off; // number of values to add
+        off = rid;
+        dst.addZeros(add - 1); // append (add-1) zeros
+        v.add2Chunk(dst);    // followed by a value
+        remain -= add;
+        assert remain >= 0;
+      }
+      // Handle case when last added value is followed by zeros till startRow+nrows
+      dst.addZeros(remain);
 
+      assert dst._len == oc._len : "NewChunk.dst.len = " + dst._len + ", oc._len = " + oc._len;
+      dst.close(dst.cidx(), fs);
+    } catch(RuntimeException t){
+      Log.err("gor exception in chunkSplitter, ic = " + ic + ", oc = " + oc + " startRow = " + startRow + " nrows = " + nrows);
+      throw t;
+    }
     return ;
   }
 }
diff --git a/src/main/java/water/fvec/NewChunk.java b/src/main/java/water/fvec/NewChunk.java
@@ -3,6 +3,7 @@
 import java.util.*;
 
 import water.*;
+import water.util.Log;
 
 // An uncompressed chunk of data, supporting an append operation
 public class NewChunk extends Chunk {
@@ -77,31 +78,59 @@ public void add2Chunk(NewChunk c){
   }
 
   public Iterator<Value> values(int fromIdx, int toIdx){
-    final int lId, gId;
-    final int to = Math.min(toIdx,_len);
-
-    if(sparse()){
-      int x = Arrays.binarySearch(_id,0,_sparseLen,fromIdx);
-      if(x < 0) x = -x -1;
-      lId = x;
-      gId = x == _sparseLen?_len:_id[x];
-    } else
-      lId = gId = fromIdx;
-    final Value v = new Value(lId,gId);
-    final Value next = new Value(lId,gId);
-    return new Iterator<Value>(){
-      @Override public final boolean hasNext(){return next._gId < to;}
-      @Override public final Value next(){
-        if(!hasNext())throw new NoSuchElementException();
-        v._gId = next._gId; v._lId = next._lId;
-        next._lId++;
-        if(sparse()) next._gId = next._lId < _sparseLen?_id[next._lId]:_len;
-        else next._gId++;
-        return v;
-      }
-      @Override
-      public void remove() {throw new UnsupportedOperationException();}
-    };
+    try {
+      final int lId, gId;
+      final int to = Math.min(toIdx, _len);
+
+      if (sparse()) {
+        int x = Arrays.binarySearch(_id, 0, _sparseLen, fromIdx);
+        if (x < 0) x = -x - 1;
+        lId = x;
+        gId = x == _sparseLen ? _len : _id[x];
+      } else
+        lId = gId = fromIdx;
+      final Value v = new Value(lId, gId);
+      final Value next = new Value(lId, gId);
+      return new Iterator<Value>() {
+        @Override
+        public final boolean hasNext() {
+          return next._gId < to;
+        }
+
+        @Override
+        public final Value next() {
+          if (!hasNext()) throw new NoSuchElementException();
+          v._gId = next._gId;
+          v._lId = next._lId;
+          next._lId++;
+          if (sparse()) next._gId = next._lId < _sparseLen ? _id[next._lId] : _len;
+          else next._gId++;
+          return v;
+        }
+
+        @Override
+        public void remove() {
+          throw new UnsupportedOperationException();
+        }
+      };
+    }catch(RuntimeException t){
+      try {
+        StringBuilder sb = new StringBuilder("NewChunk: got exception during values() call, _len = " + _len + " _sparseLen = " + _sparseLen + " _isSparse = " + isSparse() + ", isDouble = " + (_ds != null) + "\n");
+        // print first 10 elems
+        for (int i = 0; i < Math.min(len(),10); ++i)
+          sb.append(i + ": rowId = " + (_id == null ? i : _id[i]) + ", value = " + (_ds == null ? (_ds[i]) : (_ls[i] + " e" + _xs[i])) + "\n");
+        // print last 10
+        if(len() > 10) {
+          sb.append("...");
+          for(int i = Math.max(10,len()-10); i < len(); ++i)
+            sb.append(i + ": rowId = " + (_id == null ? i : _id[i]) + ", value = " + (_ds == null ? (_ds[i]) : (_ls[i] + " e" + _xs[i])) + "\n");
+        }
+        Log.err(sb.toString());
+      } catch(Throwable tt){
+        Log.err(tt);
+      } // just in case there is a bug in my printout, don't mask original exception!
+      throw t;
+    }
   }
 
 
@@ -467,6 +496,8 @@ protected void cancel_sparse(){
 
   Chunk compress() {
     Chunk res = compress2();
+    assert _len == res.len();
+    assert !sparse() || !res.isSparse() || sparseLen() == res.sparseLen();
     // force everything to null after compress to free up the memory
     _id = null;
     _xs = null;
diff --git a/src/main/java/water/fvec/RebalanceDataSet.java b/src/main/java/water/fvec/RebalanceDataSet.java
@@ -4,6 +4,7 @@
 import water.H2O;
 import water.Key;
 import water.MRTask2;
+import water.util.Log;
 
 import java.util.Arrays;
 import java.util.Iterator;
@@ -98,41 +99,46 @@ public static class RebalanceTask extends MRTask2<RebalanceTask> {
     @Override public boolean logVerbose() { return false; }
 
     private void rebalanceChunk(Vec srcVec, Chunk chk){
-      NewChunk dst = new NewChunk(chk);
-      dst._len = dst._sparseLen = 0;
-      int rem = chk._len;
-      while(rem > 0 && dst._len < chk._len){
-        Chunk srcRaw = srcVec.chunkForRow(chk._start+dst._len);
-        NewChunk src = new NewChunk((srcRaw));
-        src = srcRaw.inflate_impl(src);
-        assert src._len == srcRaw._len;
-        int srcFrom = (int)(chk._start+dst._len - src._start);
-        // check if the result is sparse (not exact since we only take subset of src in general)
-        if((src.sparse() && dst.sparse()) || (src._len + dst._len < NewChunk.MIN_SPARSE_RATIO*(src._len + dst._len))){
-          src.set_sparse(src._sparseLen);
-          dst.set_sparse(dst._sparseLen);
+      try {
+        NewChunk dst = new NewChunk(chk);
+        dst._len = dst._sparseLen = 0;
+        int rem = chk._len;
+        while (rem > 0 && dst._len < chk._len) {
+          Chunk srcRaw = srcVec.chunkForRow(chk._start + dst._len);
+          NewChunk src = new NewChunk((srcRaw));
+          src = srcRaw.inflate_impl(src);
+          assert src._len == srcRaw._len;
+          int srcFrom = (int) (chk._start + dst._len - src._start);
+          // check if the result is sparse (not exact since we only take subset of src in general)
+          if ((src.sparse() && dst.sparse()) || (src._len + dst._len < NewChunk.MIN_SPARSE_RATIO * (src._len + dst._len))) {
+            src.set_sparse(src._sparseLen);
+            dst.set_sparse(dst._sparseLen);
+          }
+          final int srcTo = srcFrom + rem;
+          int off = srcFrom - 1;
+          Iterator<NewChunk.Value> it = src.values(Math.max(0, srcFrom), srcTo);
+          while (it.hasNext()) {
+            NewChunk.Value v = it.next();
+            final int rid = v.rowId0();
+            assert rid < srcTo;
+            int add = rid - off;
+            off = rid;
+            dst.addZeros(add - 1);
+            v.add2Chunk(dst);
+            rem -= add;
+            assert rem >= 0;
+          }
+          int trailingZeros = Math.min(rem, src._len - off - 1);
+          dst.addZeros(trailingZeros);
+          rem -= trailingZeros;
         }
-        final int srcTo = srcFrom + rem;
-        int off = srcFrom-1;
-        Iterator<NewChunk.Value> it = src.values(Math.max(0,srcFrom),srcTo);
-        while(it.hasNext()){
-          NewChunk.Value v = it.next();
-          final int rid = v.rowId0();
-          assert  rid < srcTo;
-          int add = rid - off;
-          off = rid;
-          dst.addZeros(add-1);
-          v.add2Chunk(dst);
-          rem -= add;
-          assert rem >= 0;
-        }
-        int trailingZeros = Math.min(rem,src._len - off -1);
-        dst.addZeros(trailingZeros);
-        rem -= trailingZeros;
+        assert rem == 0 : "rem = " + rem;
+        assert dst._len == chk._len : "len2 = " + dst._len + ", _len = " + chk._len;
+        dst.close(dst.cidx(), _fs);
+      } catch(RuntimeException t){
+        Log.err("got exception while rebalancing chunk " + chk);
+        throw t;
       }
-      assert rem == 0:"rem = " + rem;
-      assert dst._len == chk._len:"len2 = " + dst._len + ", _len = " + chk._len;
-      dst.close(dst.cidx(),_fs);
     }
     @Override public void map(Chunk [] chks){
       for(int i = 0; i < chks.length; ++i)
