Reworked serialization api

Author: cogmission

.classpath

@@ -18,10 +18,7 @@
 	<classpathentry kind="lib" path="libs/slf4j-api-1.7.10.jar"/>
 	<classpathentry kind="lib" path="libs/trove-3.1a1.jar" sourcepath="/Users/cogmission/Resources/trove/3.1a1/trove-3.1a1-src.jar"/>
 	<classpathentry kind="lib" path="libs/java-util-1.19.3.jar" sourcepath="/Users/cogmission/.m2/repository/com/cedarsoftware/java-util/1.19.3/java-util-1.19.3-sources.jar"/>
-	<classpathentry kind="lib" path="libs/kryo-shaded-3.0.3.jar" sourcepath="/Users/cogmission/.m2/repository/com/esotericsoftware/kryo-shaded/3.0.3/kryo-shaded-3.0.3-sources.jar"/>
-	<classpathentry kind="lib" path="libs/kryo-serializers-0.37.jar" sourcepath="/Users/cogmission/.m2/repository/de/javakaffee/kryo-serializers/0.37/kryo-serializers-0.37-sources.jar"/>
 	<classpathentry kind="lib" path="libs/objenesis-2.1.jar"/>
-	<classpathentry kind="lib" path="libs/minlog-1.3-SNAPHOT.jar"/>
 	<classpathentry combineaccessrules="false" kind="src" path="/fast-serialization"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>

Changelog.txt

@@ -19,5 +19,3 @@ Changes for Serialization branch (notes for CHANGELOG.md)
 11. Added NetworkSerializer interface (and NetworkSerializerImpl) which executes FST serialization
     and simultaneously implements the Kryo interface also.
 12. Added the SerialConfig class which contains config information for the NetworkSerializer.
-13. Added Network.Scheme enum which has two values { FST, KRYO }, to designate underlying 
-    serialization scheme (Kryo still delegates to FST but implements Kryo Serializer interface)

build.gradle

@@ -49,9 +49,6 @@ dependencies {
     compile group: 'org.slf4j', name: 'slf4j-api', version:'1.7.10'
 	compile group: 'io.reactivex', name: 'rxjava', version: '1.0.10'
 	compile group: 'de.ruedigermoeller', name: 'fst', version: '2.45'
-	compile group: 'com.esotericsoftware', name: 'kryo-shaded', version: '3.0.3'
-	compile group: 'com.esotericsoftware', name: 'minlog', version: '1.3-SNAPSHOT'
-	compile group: 'de.javakaffee', name: 'kryo-serializers', version: '0.37'
 	compile group: 'com.cedarsoftware', name: 'java-util', version: '1.19.3'
 	testCompile group: 'junit', name: 'junit', version:'4.11'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version:'1.1.2'

libs/kryo-serializers-0.37.jar

@@ -49,8 +49,8 @@
 			<groupId>com.chaschev</groupId>
 			<artifactId>chutils</artifactId>
 			<version>1.4</version>
-			<!--We're only interested in ASCIITable from this library, so exclude all 
-				transitive dependencies -->
+			<!--We're only interested in ASCIITable from this library, so exclude 
+				all transitive dependencies -->
 			<exclusions>
 				<exclusion>
 					<groupId>org.mockito</groupId>
@@ -128,21 +128,6 @@
 			<version>2.45</version>
 		</dependency>
 		<dependency>
-			<groupId>com.esotericsoftware</groupId>
-			<artifactId>kryo-shaded</artifactId>
-			<version>3.0.3</version>
-		</dependency>
-		<dependency>
-			<groupId>com.esotericsoftware</groupId>
-			<artifactId>minlog</artifactId>
-			<version>1.3-SNAPHOT</version>
-		</dependency>
-		<dependency>
-			<groupId>de.javakaffee</groupId>
-			<artifactId>kryo-serializers</artifactId>
-			<version>0.37</version>
-        </dependency>
-        <dependency>
 			<groupId>org.openjdk.jmh</groupId>
 			<artifactId>jmh-core</artifactId>
 			<version>1.11.3</version>

libs/kryo-shaded-3.0.3.jar

@@ -204,6 +204,9 @@ public void burstColumns(Connections c, ComputeCycle cycle,
         activeColumns.removeAll(predictedColumns);
 
         for(Column column : activeColumns) {
+            if(column == null || column.getCells() == null) {
+                System.out.println("here");
+            }
             List<Cell> cells = column.getCells();
             cycle.activeCells.addAll(cells);
 

libs/minlog-1.3-SNAPHOT.jar

@@ -28,32 +28,33 @@
  * <p>
  * Executes check point behavior through the {@link #checkPoint(Observer)} method. The
  * checkPoint() method adds the specified {@link rx.Observer} to the list of those
- * observers notified following a check point operation. The notification consists of
+ * observers notified following a check point operation. This "subscribe" action invokes
+ * the underlying check point operation and returns a notification. The notification consists of
  * a byte[] containing the serialized {@link Network}.
  * </p><p>
  * <b>Typical usage is as follows:</b>
  * <pre>
- *  {@link Network} network = ...
+ *  {@link Persistence} p = Persistence.get();
  *  
- *  network.checkPointer().checkPoint(new Observer<byte[]>() { 
+ *  p.checkPointOp().checkPoint(new Observer<byte[]>() { 
  *      public void onCompleted() {}
  *      public void onError(Throwable e) { e.printStackTrace(); }
  *      public void onNext(byte[] bytes) {
  *          // Do work here, use serialized Network byte[] here if desired...
  *      }
  *  });
  * 
- * Merely by adding this Observer, the Network knows to check point after completion of 
+ * Again, by subscribing to this CheckPointOp, the Network knows to check point after completion of 
  * the current compute cycle (it checks the List of Observers to see if it's non-empty).
  * Then after it notifies all current observers, it clears the list prior to the next 
- * following compute cycle. see {@link PersistenceAPI} for a more detailed discussion...
+ * following compute cycle. see {@link PAPI} for a more detailed discussion...
  * 
  * @author cogmission
  *
- * @param <T>  a {@link Network} 
+ * @param <T>  the notification return type
  */
 @FunctionalInterface
-public interface CheckPointer<T> {
+public interface CheckPointOp<T> {
     /**
      * Registers the Observer for a single notification following the checkPoint
      * operation. The user will be notified with the byte[] of the {@link Network}

pom.xml

@@ -226,8 +226,8 @@ public class Layer<T> implements Persistable {
     private Layer<Inference> previous;
 
     private transient List<Observer<Inference>> observers = new ArrayList<Observer<Inference>>();
-    private transient CheckPointer<byte[]> checkPointer;
-    private transient List<Observer<byte[]>> checkPointObservers = new ArrayList<>();
+    private transient CheckPointOperator<?> checkPointOp;
+    private transient List<Observer<byte[]>> checkPointOpObservers = new ArrayList<>();
 
 
 
@@ -325,7 +325,7 @@ public Layer<T> postDeSerialize() {
         factory = new FunctionFactory();
         factory.inference = old.inference.postSerialize(old.inference);
 
-        checkPointObservers = new ArrayList<>();
+        checkPointOpObservers = new ArrayList<>();
 
         if(sensor != null) {
             sensor.setLocalParameters(params);
@@ -404,7 +404,18 @@ public Layer(Parameters params, MultiEncoder e, SpatialPooler sp, TemporalMemory
                 (anomalyComputer == null ? "" : "Anomaly"));
         }
     }
-
+    
+    /**
+     * USED INTERNALLY, DO NOT CALL.
+     * @return
+     */
+    public CheckPointOp<byte[]> delegateCheckPointCall() {
+        if(parentNetwork != null) {
+            return parentNetwork.getCheckPointOperator();
+        }
+        return null;
+    }
+    
     /**
      * Sets the parent region which contains this {@code Layer}
      * 
@@ -1426,9 +1437,9 @@ public int getMostProbableBucketIndex(String field, int step) {
         return c.getMostProbableBucketIndex(step);
     }
 
-    // ////////////////////////////////////////////////////////////
-    // PRIVATE METHODS AND CLASSES BELOW HERE //
-    // ////////////////////////////////////////////////////////////
+    //////////////////////////////////////////////////////////////
+    //          PRIVATE METHODS AND CLASSES BELOW HERE          //
+    //////////////////////////////////////////////////////////////
     /**
      * Notify all subscribers through the delegate that stream processing has
      * been completed or halted.
@@ -1639,7 +1650,7 @@ private Observable<ManualInput> resolveObservableSequence(T t) {
         }
 
         sequenceStart = sequenceStart.filter(m -> {
-            if(!checkPointObservers.isEmpty() && parentNetwork != null) {
+            if(!checkPointOpObservers.isEmpty() && parentNetwork != null) {
                 // Execute check point logic
                 doCheckPoint();
             }
@@ -1656,21 +1667,21 @@ private Observable<ManualInput> resolveObservableSequence(T t) {
      * Observers; then clears the list of Observers.
      */
     private void doCheckPoint() {
-        byte[] bytes = parentNetwork.internalCheckPoint();
+        byte[] bytes = parentNetwork.internalCheckPointOp();
 
         if(bytes != null) {
             LOGGER.debug("Layer [" + getName() + "] checkPointed file: " + 
-                Network.serializer(null, false).getLastCheckPointFileName() + ",  hc = " + Network.serializer(null, false).hashCode());
+                Persistence.get().getLastCheckPointFileName());
         }else{
             LOGGER.debug("Layer [" + getName() + "] checkPoint   F A I L E D   at: " + (new DateTime()));
         }
 
-        for(Observer<byte[]> o : checkPointObservers) {
+        for(Observer<byte[]> o : checkPointOpObservers) {
             o.onNext(bytes);
             o.onCompleted();
         }
 
-        checkPointObservers.clear();
+        checkPointOpObservers.clear();
     }
 
     /**
@@ -2005,25 +2016,29 @@ public void run() {
     }
 
     /**
-     * Returns the pre-built subscribe function used to add subscribers (callers of 
-     * {@link #checkPointer()}) to the check point observer notifications.
-     *  
-     * @return  the internal Observable used for post check point notifications.
+     * Returns an {@link rx.Observable} operator that when subscribed to, invokes an operation
+     * that stores the state of this {@code Network} while keeping the Network up and running.
+     * The Network will be stored at the pre-configured location (in binary form only, not JSON).
+     * 
+     * @param network   the {@link Network} to check point.
+     * @return  the {@link CheckPointOp} operator 
      */
-    CheckPointer<byte[]> checkPointer() {
-        if(checkPointer == null) {
-            checkPointer = new CheckPointerImpl<byte[]>(Layer.this);
+    @SuppressWarnings("unchecked")
+    CheckPointOp<byte[]> getCheckPointOperator() {
+        if(checkPointOp == null) {
+            checkPointOp = new CheckPointOperator<byte[]>(Layer.this);
         }
-        return checkPointer;
+        return (CheckPointOp<byte[]>)checkPointOp;
     }
 
 
     //////////////////////////////////////////////////////////////
     //   Inner Class Definition for CheckPointer (Observable)   //
     //////////////////////////////////////////////////////////////
+    
     /**
      * <p>
-     * Implementation of the CheckPointer interface which serves to checkpoint
+     * Implementation of the CheckPointOp interface which serves to checkpoint
      * and register a listener at the same time. The {@link rx.Observer} will be
      * notified with the byte array of the {@link Network} being serialized.
      * </p><p>
@@ -2032,18 +2047,18 @@ CheckPointer<byte[]> checkPointer() {
      * be executed.
      * </p>
      * 
-     * @param <T>       {@link rx.Observer} 
+     * @param <T>       {@link rx.Observer}'s return type
      */
-    static class CheckPointerImpl<T> extends Observable<T> implements CheckPointer<T> {
-        private CheckPointerImpl(Layer<?> l) {
+    static class CheckPointOperator<T> extends Observable<T> implements CheckPointOp<T> {
+        private CheckPointOperator(Layer<?> l) {
             this(new Observable.OnSubscribe<T>() {
-                @SuppressWarnings("unchecked")
-                @Override public void call(Subscriber<? super T> t) {
+                @SuppressWarnings({ "unchecked" })
+                @Override public void call(Subscriber<? super T> r) {
                     if(l.LAYER_THREAD != null) {
                         // The layer thread automatically tests for the list of observers to 
                         // contain > 0 elements, which indicates a check point operation should
                         // be executed.
-                        l.checkPointObservers.add((Observer<byte[]>)t);
+                        l.checkPointOpObservers.add((Observer<byte[]>)r);
                     }else{
                         l.doCheckPoint();
                     }
@@ -2052,10 +2067,10 @@ private CheckPointerImpl(Layer<?> l) {
         }
 
         /**
-         * Constructs this {@code CheckPointerImpl}
+         * Constructs this {@code CheckPointOperator}
          * @param f     a subscriber function
          */
-        protected CheckPointerImpl(rx.Observable.OnSubscribe<T> f) {
+        protected CheckPointOperator(rx.Observable.OnSubscribe<T> f) {
             super(f);
         }