* Licensed to the Apache Software Foundation (ASF) under one

* or more contributor license agreements.  See the NOTICE file

* distributed with this work for additional information

* regarding copyright ownership.  The ASF licenses this file

* to you under the Apache License, Version 2.0 (the

* "License"); you may not use this file except in compliance

* with the License.  You may obtain a copy of the License at

*

*   http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing,

* software distributed under the License is distributed on an

* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

* KIND, either express or implied.  See the License for the

* specific language governing permissions and limitations

* under the License.

*/

* User callable serialization and deserialization of the CAS in a compressed Binary Format

*

* This serializes/deserializes the state of the CAS, assuming that the type

* information remains constant.

*

* Header specifies to reader the format, and the compression level.

*

* How to Serialize: 

*

* 1) create an instance of this class, specifying some options that don't change very much

* 2) call serialize(CAS) to serialize the cas *

*

* You can reuse the instance for a different CAS (as long as the type system is the same);

* this will save setup time.

*

* This class lazily constructs customized TypeInfo instances for each type encountered in serializing.

* These are preserved across multiple serialization calls, so their setup / initialization is only

* needed the first time.

*

* The form of the binary CAS is inserted at the beginning so that receivers can do the

* proper deserialization. 

*    

* Binary format requires that the exact same type system be used when deserializing

*

* How to Deserialize:

*

* 1) get an appropriate CAS to deserialize into.  For delta CAS, it does not have to be empty.

* 2) call CASImpl: cas.reinit(inputStream)  This is the existing method

*    for binary deserialization, and it now handles this compressed version, too.

*    Delta cas is also supported.

*

* Compression/Decompression

* Works in two stages:

*   application of Zip/Unzip to particular sub-collections of CAS data,

*     grouped according to similar data distribution

*   collection of like kinds of data (to make the zipping more effective)

*   There can be up to ~20 of these collections, such as

*      control info, float-exponents, string chars

* Deserialization:

*   Read all bytes,

*   create separate ByteArrayInputStreams for each segment, sharing byte bfr

*   create appropriate unzip data input streams for these

*/

 

  public static final int TYPECODE_COMPR = 8;

  public static final boolean CHANGE_FS_REFS_TO_SEQUENTIAL = true;

  // may add more later - to specify differing trade-offs between speed and compression

  public enum Compression {None, Compress}; 

  public static final boolean IS_DIFF_ENCODE = true;

  public static final boolean CAN_BE_NEGATIVE = true;

  public static final boolean IGNORED = true;

  public static final boolean IN_MAIN_HEAP = true;

 

  private static final long DBL_1 = Double.doubleToLongBits(1D);

  /**

   * The kinds of slots that can exist

   *   an index for getting type-code specific values,

   *   flag - whether or not they should be diff encoded

   *   flag - if they can be negative (and need their sign moved)

   *  

   * Some are real slots in the heap; others are descriptions of

   *   parts of values, eg. float exponent

   *  

   * Difference encoding costs 1 bit. 

   *   Measurements show it can lessen zip's effectiveness

   *     (especially for single byte values (?)),

   *     probably because it causes more dispersion in

   *     the value kinds. 

   *   Because of this 2-fold cost (1 bit and less zip),

   *     differencing being tried only for multi-byte

   *     values (short, int, long), and heap refs

   *     - for array values, diff is with prev array value

   *       (for 1st value in array, diff is with prev FeatureStructure

   *       of the same type in the heap's 1st value if it exists

   *     - for non-array values or 1st array value, diff is with

   *       prev heap value for same type in heap 

   *

   *   Not done for float parts - exponent too short, and

   *     mantissa too random.

   *

   * CanBeNegative

   *   Many values are only positive e.g., array lengths

   *   Some values can be negative

   *     (all difference-encoded things can be negative)

   *   Represent as 1 bit + positive number, sign bit in

   *     least sig. bit position.  This allows the

   *     bits to cluster closer to 0 on the positive side,

   *     which can make for fewer bytes to represent the number.

   */

  /**

   * Compression alternatives

   */

 

  public enum CompressLevel {

    None(   Deflater.NO_COMPRESSION),

    Fast(   Deflater.BEST_SPEED),

    Default(Deflater.DEFAULT_COMPRESSION),

    Best(   Deflater.BEST_COMPRESSION),

    ;

    final public int lvl;

    CompressLevel(int lvl) {

      this.lvl = lvl;

    }

  }

 

  public enum CompressStrat {

    Default(      Deflater.DEFAULT_STRATEGY),

    Filtered(     Deflater.FILTERED),

    HuffmanOnly(  Deflater.HUFFMAN_ONLY),

    ;

    final public int strat;

    CompressStrat(int strat) {

      this.strat = strat;

    }

  }

 

  /**

   * Define all the slot kinds.

   */

  public enum SlotKind {

    Slot_ArrayLength(! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, IN_MAIN_HEAP),

    Slot_HeapRef(    IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_Int(        IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_Byte(       ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, IN_MAIN_HEAP),

    Slot_Short(      IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_TypeCode(   ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, IN_MAIN_HEAP),

    Slot_StrOffset(  ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, !IN_MAIN_HEAP),

    Slot_StrLength(  ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, !IN_MAIN_HEAP),

    Slot_Long_High(    IS_DIFF_ENCODE,           IGNORED, 0, !IN_MAIN_HEAP),

    Slot_Long_Low (    IS_DIFF_ENCODE,           IGNORED, 0, !IN_MAIN_HEAP),

    // the next are not actual slot kinds, but instead

    // are codes used to control encoding of Floats and Doubles.

    Slot_Float_Mantissa_Sign( ! IS_DIFF_ENCODE, CAN_BE_NEGATIVE, 0, !IN_MAIN_HEAP),

    // exponent is 8 bits, and shifted in the expectation

    // that many values may be between 1 and 0 (e.g., normalized values)

    //   -- so sign moving is needed

    Slot_Float_Exponent(      ! IS_DIFF_ENCODE, CAN_BE_NEGATIVE, 0, !IN_MAIN_HEAP),

   

    Slot_Double_Mantissa_Sign(! IS_DIFF_ENCODE, CAN_BE_NEGATIVE, 0, !IN_MAIN_HEAP),

    Slot_Double_Exponent(     ! IS_DIFF_ENCODE, CAN_BE_NEGATIVE, 0, !IN_MAIN_HEAP),

    Slot_FsIndexes(             IS_DIFF_ENCODE,         IGNORED, 4, !IN_MAIN_HEAP),

   

    Slot_StrChars(            IGNORED,          IGNORED, 2, !IN_MAIN_HEAP),

   

    Slot_Control(             IGNORED,          IGNORED, 0, !IN_MAIN_HEAP),

    Slot_StrSeg(              ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 0, ! IN_MAIN_HEAP),

   

    // the next slots are not serialized

    Slot_StrRef(     IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_BooleanRef( ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, IN_MAIN_HEAP),

    Slot_ByteRef(    IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_ShortRef(   IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_LongRef(    IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_DoubleRef(  IS_DIFF_ENCODE,             IGNORED, 4, IN_MAIN_HEAP),

    Slot_Float(      ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, IN_MAIN_HEAP),

    Slot_Boolean(    ! IS_DIFF_ENCODE, ! CAN_BE_NEGATIVE, 4, IN_MAIN_HEAP),

    // next used to capture original heap size

    Slot_MainHeap(   IGNORED,          IGNORED,           4, !IN_MAIN_HEAP),

    ;

    public final int i;

    public final boolean isDiffEncode;

    public final boolean canBeNegative;

    public final boolean inMainHeap;

    public final int elementSize;

   

    public static final int NBR_SLOT_KIND_ZIP_STREAMS;

    static {NBR_SLOT_KIND_ZIP_STREAMS = Slot_StrRef.i;}

   

    SlotKind(boolean isDiffEncode,

             boolean canBeNegative,

             int elementSize,

             boolean inMainHeap) {

      this.i = this.ordinal();

      this.isDiffEncode = isDiffEncode;

      this.canBeNegative = isDiffEncode ? true : canBeNegative;

      this.elementSize = elementSize;

      this.inMainHeap = inMainHeap;

    }

  }

 

  /**

   * Things set up for one instance of this class, and

   * reuse-able

   */

  final private TypeInfo [] typeInfoArray;  // lazy initialization of elements

  final private TypeSystemImpl ts;  // for debugging

  final private boolean doMeasurements;

 

  // speedups

  final private static int arrayLength_i = Slot_ArrayLength.i;

  final private static int heapRef_i = Slot_HeapRef.i;

  final private static int int_i = Slot_Int.i;

  final private static int byte_i = Slot_Byte.ordinal();

  final private static int short_i = Slot_Short.i;

  final private static int typeCode_i = Slot_TypeCode.i;

  final private static int strOffset_i = Slot_StrOffset.i;

  final private static int strLength_i = Slot_StrLength.i;

  final private static int long_High_i = Slot_Long_High.i;

  final private static int long_Low_i = Slot_Long_Low.i;

  final private static int float_Mantissa_Sign_i = Slot_Float_Mantissa_Sign.i;

  final private static int float_Exponent_i = Slot_Float_Exponent.i;

  final private static int double_Mantissa_Sign_i = Slot_Double_Mantissa_Sign.i;

  final private static int double_Exponent_i = Slot_Double_Exponent.i;

  final private static int fsIndexes_i = Slot_FsIndexes.i;

  final private static int strChars_i = Slot_StrChars.i;

  final private static int control_i = Slot_Control.i;

  final private static int strSeg_i = Slot_StrSeg.i;

 

  /**

   *

   * @param ts

   * @param doMeasurements - normally set this to false.

   */

  public BinaryCasSerDes4(TypeSystemImpl ts, boolean doMeasurements) {

    this.ts = ts;

    this.doMeasurements = doMeasurements;

    typeInfoArray = new TypeInfo[(ts.getTypeArraySize())];

  }

  /**

   *

   * @param cas

   * @param out

   * @param trackingMark

   * @return null or serialization measurements (depending on setting of doMeasurements)

   * @throws IOException

   */

  public SerializationMeasures serialize(AbstractCas cas, Object out, Marker trackingMark,

      CompressLevel compressLevel, CompressStrat compressStrategy) throws IOException {

    SerializationMeasures sm = (doMeasurements) ? new SerializationMeasures() : null;

    CASImpl casImpl = (CASImpl) ((cas instanceof JCas) ? ((JCas)cas).getCas(): cas);

    if (null != trackingMark && !trackingMark.isValid() ) {

      throw new CASRuntimeException(

                CASRuntimeException.INVALID_MARKER, new String[] { "Invalid Marker." });

    }

   

    Serializer serializer = new Serializer(

        casImpl, makeDataOutputStream(out), (MarkerImpl) trackingMark, sm,

        compressLevel, compressStrategy);

  

    serializer.serialize();

    return sm;

  }

 

  public SerializationMeasures serialize(AbstractCas cas, Object out, Marker trackingMark,

      CompressLevel compressLevel) throws IOException {

    return serialize(cas, out,trackingMark, compressLevel, CompressStrat.Default);

  }

 

  public SerializationMeasures serialize(AbstractCas cas, Object out, Marker trackingMark) throws IOException {

    return serialize(cas, out,trackingMark, CompressLevel.Default, CompressStrat.Default);

  }

  public SerializationMeasures serialize(AbstractCas cas, Object out) throws IOException {

    return serialize(cas, out, null);

  }

  public void deserialize(CASImpl cas, InputStream deserIn, boolean isDelta) throws IOException {

    DataInput in;

    if (deserIn instanceof DataInputStream) {

      in = (DataInputStream)deserIn;

    } else {

      in = new DataInputStream(deserIn);

    }

    Deserializer deserializer = new Deserializer(cas, in, isDelta);   

    deserializer.deserialize();

  }

  /**

   * Class instantiated once per serialization

   * Multiple serializations in parallel supported, with

   * multiple instances of this

   */

 

  private class Serializer {

    final private DataOutputStream serializedOut;  // where to write out the serialized result

    final private CASImpl cas;  // cas being serialized

    final private MarkerImpl mark;  // the mark to serialize from

   

    final private SerializationMeasures sm;  // null or serialization measurements

    final private ByteArrayOutputStream[] baosZipSources = new ByteArrayOutputStream[NBR_SLOT_KIND_ZIP_STREAMS];  // lazily created, indexed by SlotKind.i

    final private DataOutputStream[] dosZipSources = new DataOutputStream[NBR_SLOT_KIND_ZIP_STREAMS];      // lazily created, indexed by SlotKind.i

    final private int[] heap;           // main heap

    private int heapStart;

    final private int heapEnd;

    final private StringHeap stringHeapObj;

    final private LongHeap longHeapObj;

    final private ShortHeap shortHeapObj;

    final private ByteHeap byteHeapObj;

    final private boolean isDelta;        // if true, there is a marker indicating the start spot(s)

    final private boolean doMeasurement;  // if true, doing measurements

    final private ComprItemRefs fsStartIndexes = (CHANGE_FS_REFS_TO_SEQUENTIAL) ? new ComprItemRefs() : null;

    final private int[] typeCodeHisto = new int[ts.getTypeArraySize()];

    final private Integer[] serializedTypeCode2Code = new Integer[ts.getTypeArraySize()]; // needs to be Integer to get comparator choice

    final private int[] estimatedZipSize = new int[NBR_SLOT_KIND_ZIP_STREAMS]; // one entry for each output stream kind

    final private OptimizeStrings os;

    final private CompressLevel compressLevel;

    final private CompressStrat compressStrategy;

   

    // typeInfo is local to this serialization instance to permit multiple threads

    private TypeInfo typeInfo; // type info for the current type being serialized

    private int iPrevHeap;        // 0 or heap addr of previous instance of current type

    private boolean only1CommonString;  // true if only one common string

   

    // speedups

   

    // any use of these means caller handles measurement

    // some of these are never used, because the current impl

    //   is using the _i form to get measurements done

    final private DataOutputStream arrayLength_dos;

    final private DataOutputStream heapRef_dos;

    final private DataOutputStream int_dos;

    final private DataOutputStream byte_dos;

    final private DataOutputStream short_dos;

    final private DataOutputStream typeCode_dos;

    final private DataOutputStream strOffset_dos;

    final private DataOutputStream strLength_dos;

    final private DataOutputStream long_High_dos;

    final private DataOutputStream long_Low_dos;

    final private DataOutputStream float_Mantissa_Sign_dos;

    final private DataOutputStream float_Exponent_dos;

    final private DataOutputStream double_Mantissa_Sign_dos;

    final private DataOutputStream double_Exponent_dos;

    final private DataOutputStream fsIndexes_dos;

    final private DataOutputStream strChars_dos;

    final private DataOutputStream control_dos;

    final private DataOutputStream strSeg_dos;

    private Serializer(CASImpl cas, DataOutputStream serializedOut, MarkerImpl mark,

                       SerializationMeasures sm,

                       CompressLevel compressLevel,

                       CompressStrat compressStrategy) {

      this.cas = cas;

      this.serializedOut = serializedOut;

      this.mark = mark;

      this.sm = sm;

      this.compressLevel = compressLevel;

      this.compressStrategy = compressStrategy;

      isDelta = (mark != null);

      doMeasurement = (sm != null);

     

      heap = cas.getHeap().heap;

      heapEnd = cas.getHeap().getCellsUsed();

      heapStart = isDelta ? mark.getNextFSId() : 0;

     

      stringHeapObj = cas.getStringHeap();

      longHeapObj   = cas.getLongHeap();

      shortHeapObj  = cas.getShortHeap();

      byteHeapObj   = cas.getByteHeap();

    

      os = new OptimizeStrings(doMeasurement);

     

      setupOutputStreams();  

      arrayLength_dos = dosZipSources[arrayLength_i];

      heapRef_dos = dosZipSources[heapRef_i];

      int_dos = dosZipSources[int_i];

      byte_dos = dosZipSources[byte_i];

      short_dos = dosZipSources[short_i];

      typeCode_dos = dosZipSources[typeCode_i];

      strOffset_dos = dosZipSources[strOffset_i];

      strLength_dos = dosZipSources[strLength_i];

      long_High_dos = dosZipSources[long_High_i];

      long_Low_dos = dosZipSources[long_Low_i];

      float_Mantissa_Sign_dos = dosZipSources[float_Mantissa_Sign_i];

      float_Exponent_dos = dosZipSources[float_Exponent_i];

      double_Mantissa_Sign_dos = dosZipSources[double_Mantissa_Sign_i];

      double_Exponent_dos = dosZipSources[double_Exponent_i];

      fsIndexes_dos = dosZipSources[fsIndexes_i];

      strChars_dos = dosZipSources[strChars_i];

      control_dos = dosZipSources[control_i];

      strSeg_dos = dosZipSources[strSeg_i];

    }

   

    private void setupOutputStreams() {

      // estimate model:

      //   33% of space in strings, 33% in heap, 33% other

      //   compr ratio for heap is 98%

      int compr = (heapEnd - heapStart) * 8 / 3 / 50;

      int compr1000 = Math.max(512, compr/1000);

      // 2nd arg is the number of bytes in the byte output stream, initially

      estimatedZipSize[typeCode_i] = Math.max(512, compr/4);  // /4 for ~4 slots per fs

//      estimatedZipSize[boolean_i] =compr1000;

      estimatedZipSize[byte_i] = compr1000;

      estimatedZipSize[short_i] = compr1000;

      estimatedZipSize[int_i] = Math.max(1024, compr1000);

      estimatedZipSize[arrayLength_i] = compr1000;

      estimatedZipSize[float_Mantissa_Sign_i] = compr1000;

      estimatedZipSize[float_Exponent_i] = compr1000;

      estimatedZipSize[double_Mantissa_Sign_i] = compr1000;

      estimatedZipSize[double_Exponent_i] = compr1000;

      estimatedZipSize[long_High_i] = compr1000;

      estimatedZipSize[long_Low_i] = compr1000;

      estimatedZipSize[heapRef_i] = Math.max(1024, compr1000);

      estimatedZipSize[strOffset_i] = Math.max(512, compr/4);

      estimatedZipSize[strLength_i] = Math.max(512, compr/4);

      estimatedZipSize[fsIndexes_i] = Math.max(512, compr/8);  // /4 for ~4 slots/fs, / 2 for # indexed

      estimatedZipSize[strChars_i] = Math.max(512, compr/4); // strings compress better

      estimatedZipSize[control_i] = 128;

     

      for (int i = 0; i < baosZipSources.length; i++) {

        setupOutputStream(i);

      }

    }

   

    private void serialize() throws IOException {  

      if (doMeasurement) {

        System.out.println(printCasInfo(cas));

        sm.origAuxBytes = cas.getByteHeap().getSize();

        sm.origAuxShorts = cas.getShortHeap().getSize() * 2;

        sm.origAuxLongs = cas.getLongHeap().getSize() * 8;

        sm.totalTime = System.currentTimeMillis();

      }

      /******************

       * Process Header 

       * Standardized   

       ******************/

      // encode: bits 7 6 5 4 3 2 1 0

      //                        0 0 1 = no delta, no compression

      //                        0 1 - = delta, no compression

      //                        1 d - = compression, w/wo delta

      int version = 4 | ((isDelta) ? 2 : 0);

      CASSerializer.outputVersion(version, serializedOut);

       

      serializedOut.writeInt(0);  // reserved for future version info

      if (doMeasurement) {

        sm.header = 12;

      }

          

      /**************************

       * Strings

       **************************/

      int stringHeapStart = isDelta ? mark.nextStringHeapAddr : 1;

      int stringHeapEnd = stringHeapObj.getSize();

      for (int i = stringHeapStart; i < stringHeapEnd; i++) {

        os.add(stringHeapObj.getStringForCode(i));

      }

     

      // also add in all modified strings

      // with current design, all modified strings are guaranteed

      // to be above the mark, so this code is commented out

//    addModStr:

//      if (isDelta) {

//        serializerForModifiedFSs.sortModifications();

//        final int[] mods = serializerForModifiedFSs.modifiedMainHeapAddrs;

//        final int modLen = serializerForModifiedFSs.modMainHeapAddrsLength;

//        int nextMod = 0;

//        if (modLen > 0) {

//

//          for (int iHeap = 1; iHeap < heapStart; iHeap += incrToNextFs(heap, iHeap, typeInfo)) {

//            typeInfo = getTypeInfo(heap[iHeap]);

//            FsStringRefs fsStringRefs = new FsStringRefs(typeInfo, heap, iHeap);

//            for (int nextRef = fsStringRefs.next(); nextRef >= 0; nextRef = fsStringRefs.next()) {

//              while (nextRef > mods[nextMod]) {

//                nextMod ++;

//                if (nextMod == modLen) {

//                  break addModStr;

//                }

//              }

//              if (nextRef == mods[nextMod]) {

//                os.add(stringHeapObj.getStringForCode(heap[nextRef]));

//                nextMod ++;

//                if (nextMod == modLen) {

//                  break addModStr;

//                }

//              }

//            }

//          }  

//        }

//      }

      os.optimize();

      String [] commonStrings = os.getCommonStrings();

      writeVnumber(strChars_i, commonStrings.length);

      for (int i = 0; i < commonStrings.length; i++) {

        int startPos = dosZipSources[strChars_i].size();

        DataIO.writeUTFv(commonStrings[i], dosZipSources[strChars_i]);

        // approximate histogram

        if (doMeasurement) {

          // len is utf-8 encoding

          float len = dosZipSources[strChars_i].size() - startPos;

          // if len == chars, then all got coded as 1 byte

          // if len > chars, some were utf-8 coded as 2 bytes

          float excess = (len / commonStrings[i].length()) - 1;  // excess over length 1

          int encAs2 = (int)(excess * commonStrings[i].length());

         

          // simulate histo for all the chars, as 1 or 2 byte UTF8 encoding

          sm.statDetails[strChars_i].countTotal += commonStrings[i].length(); // total chars accum

          sm.statDetails[strChars_i].c[0] = commonStrings[i].length() - encAs2;

          sm.statDetails[strChars_i].c[1] = encAs2;

          sm.statDetails[strChars_i].lengthTotal += len;  // total as UTF-8 encode

        }

      }

     

      only1CommonString = commonStrings.length == 1;

      if (doMeasurement) {

        long commonStringsLength = 0;

        sm.stringsNbrCommon = commonStrings.length;

        int r = 0;

        for (int i = 0; i < commonStrings.length; i++) {

          r += DataIO.lengthUTFv(commonStrings[i]);

          commonStringsLength += commonStrings[i].length();

        }

        sm.stringsCommonChars = r;

        sm.stringsSavedExact = os.getSavedCharsExact() * 2;

        sm.stringsSavedSubstr = os.getSavedCharsSubstr() * 2;

        sm.statDetails[strChars_i].original = os.getSavedCharsExact() * 2

                                 + os.getSavedCharsSubstr() * 2

                                 + commonStringsLength * 2;

        sm.statDetails[strLength_i].original = (stringHeapEnd - stringHeapStart) * 4;

        sm.statDetails[strOffset_i].original = (stringHeapEnd - stringHeapStart) * 4;

      }

     

      /***************************

       * Prepare to walk main heap

       ***************************/

      writeVnumber(control_dos, heapEnd - heapStart); 

      if (doMeasurement) {

        sm.statDetails[Slot_MainHeap.i].original = (1 + heapEnd - heapStart) * 4;     

      }

     

      resetIprevious();

      if (heapStart == 0) {

        heapStart = 1;  // slot 0 not serialized, it's null / 0

      }

      if (CHANGE_FS_REFS_TO_SEQUENTIAL) {

        // scan thru all fs and save their offsets in the heap

        // to allow conversion from addr to sequential fs numbers

        initFsStartIndexes(fsStartIndexes, heap, heapStart, heapEnd, typeCodeHisto);

       

        for (int i = ts.getTypeArraySize() - 1; i >= 0; i--) {

          serializedTypeCode2Code[i] = i;

        }

       

        // set typeCode2serializeCode so that the 0th element is the typeCode with the highest frequency, etc.

        Arrays.sort(serializedTypeCode2Code, 0, serializedTypeCode2Code.length, new Comparator<Integer>() {

          public int compare(Integer o1, Integer o2) {

            return (typeCodeHisto[o1] > typeCodeHisto[o2]) ? -1 :

                   (typeCodeHisto[o1] < typeCodeHisto[o2]) ? 1 : 0;

          }

        });

       

//          int tCode = serializedTypeCode2Code[i];

//          int c = typeCodeHisto[tCode];

//          if (c > 0) {

//            System.out.format("%2d %,9d instance of Type %s%n", i, c, typeInfoArray[tCode]);

//          }

//        }

       

      }

     

     

      /***************************

       * walk main heap

       ***************************/

      for (int iHeap = heapStart; iHeap < heapEnd; iHeap += incrToNextFs(heap, iHeap, typeInfo)) {

        int tCode = heap[iHeap];  // get type code     

        typeInfo = getTypeInfo(tCode);

        iPrevHeap = typeInfo.iPrevHeap;

       

        writeVnumber(typeCode_dos, tCode);

        if (typeInfo.isHeapStoredArray) {

          serializeHeapStoredArray(iHeap);

        } else if (typeInfo.isArray) {

          serializeNonHeapStoredArray(iHeap);

        } else {

          for (int i = 1; i < typeInfo.slotKinds.length + 1; i++) {

            serializeByKind(iHeap, i);

          }

        }

     

        typeInfo.iPrevHeap = iHeap;  // make this one the "prev" one for subsequent testing

        if (doMeasurement) {

          sm.statDetails[typeCode_i].incr(DataIO.lengthVnumber(tCode));

          sm.mainHeapFSs ++;

        }

      }  // end of heap walk

     

      serializeIndexedFeatureStructures();

      if (isDelta) {

        (new SerializeModifiedFSs()).serializeModifiedFSs();

      }

      collectAndZip();

     

      if (doMeasurement) {

        sm.totalTime = System.currentTimeMillis() - sm.totalTime;

      }

    }

   

   

    private void serializeIndexedFeatureStructures() throws IOException {

      int[] fsIndexes = isDelta ? cas.getDeltaIndexedFSs(mark) : cas.getIndexedFSs();

      if (doMeasurement) {

        sm.statDetails[fsIndexes_i].original = fsIndexes.length * 4 + 1;     

      }

      int nbrViews = fsIndexes[0];

      int nbrSofas = fsIndexes[1];

      writeVnumber(control_i, nbrViews);

      writeVnumber(control_i, nbrSofas);

     

      if (doMeasurement) {

        sm.statDetails[fsIndexes_i].incr(1); // an approximation - probably correct

        sm.statDetails[fsIndexes_i].incr(1);

      }

     

      int fi = 2;

      final int end1 = nbrSofas + 2;

      for (; fi < end1; fi++) {

        writeVnumber(control_i, fsIndexes[fi]);