Skip to content

Commit cf5f4e2

Browse files
cogmissioncogmission
committed
Fix tests in JavaKryoNetworkSerializationTest that weren't working; correct serializer methods
1 parent a2f6f4a commit cf5f4e2

File tree

5 files changed

+230
-80
lines changed

5 files changed

+230
-80
lines changed

‎.classpath‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,6 @@
2222
<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"/>
2323
<classpathentry kind="lib" path="libs/objenesis-2.1.jar"/>
2424
<classpathentry kind="lib" path="libs/minlog-1.3-SNAPHOT.jar"/>
25-
<classpathentry kind="lib" path="libs/fst-2.45.jar" sourcepath="/Users/cogmission/.m2/repository/de/ruedigermoeller/fst/2.45/fst-2.45-sources.jar"/>
25+
<classpathentry combineaccessrules="false" kind="src" path="/fast-serialization"/>
2626
<classpathentry kind="output" path="bin"/>
2727
</classpath>

‎src/main/java/org/numenta/nupic/network/KryoSerializer.java‎

Lines changed: 14 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,16 @@
11
package org.numenta.nupic.network;
22

3+
import java.io.Serializable;
4+
5+
import org.numenta.nupic.Persistable;
6+
import org.nustaq.serialization.FSTConfiguration;
7+
import org.nustaq.serialization.FSTObjectInput;
8+
39
import com.esotericsoftware.kryo.Kryo;
410
import com.esotericsoftware.kryo.KryoException;
511
import com.esotericsoftware.kryo.Serializer;
612
import com.esotericsoftware.kryo.io.Input;
713
import com.esotericsoftware.kryo.io.Output;
8-
import org.numenta.nupic.Persistable;
9-
import org.nustaq.serialization.FSTConfiguration;
10-
import org.nustaq.serialization.FSTObjectInput;
11-
import org.nustaq.serialization.FSTObjectOutput;
12-
13-
import java.io.IOException;
14-
import java.io.ObjectInputStream;
15-
import java.io.Serializable;
1614

1715
/**
1816
* Primitive Kryo serializer based on FST.
@@ -25,22 +23,18 @@
2523
* Use instead {@link Network#serializer(SerialConfig, boolean)}
2624
*/
2725
public class KryoSerializer<T> extends Serializer<T> implements Serializable {
26+
private static final long serialVersionUID = 1L;
2827

29-
private final Class[] classes;
28+
private final Class<?>[] classes;
3029

3130
/** Use of Fast Serialize https://github.com/RuedigerMoeller/fast-serialization */
3231
private transient FSTConfiguration fastSerialConfig;
3332

34-
public KryoSerializer(Class... c) {
33+
public KryoSerializer(Class<?>... c) {
3534
this.classes = c;
3635
initFST();
3736
}
3837

39-
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
40-
in.defaultReadObject();
41-
initFST();
42-
}
43-
4438
private void initFST() {
4539
fastSerialConfig = FSTConfiguration.createDefaultConfiguration();
4640
fastSerialConfig.registerClass(classes);
@@ -54,15 +48,15 @@ private void initFST() {
5448
*/
5549
@Override
5650
public void write(Kryo kryo, Output output, T t) {
57-
FSTObjectOutput writer = fastSerialConfig.getObjectOutput(output);
5851
try {
5952
if(t instanceof Persistable) {
6053
((Persistable) t).preSerialize();
6154
}
62-
63-
writer.writeObject(t, t.getClass());
55+
56+
byte[] bytes = fastSerialConfig.asByteArray(t);
57+
output.write(bytes);
6458
}
65-
catch(IOException e) {
59+
catch(Exception e) {
6660
throw new KryoException(e);
6761
}
6862
}
@@ -74,6 +68,7 @@ public void write(Kryo kryo, Output output, T t) {
7468
* @param aClass The class of the object to be read in.
7569
* @return an instance of type &lt;T&gt;
7670
*/
71+
@SuppressWarnings("unchecked")
7772
@Override
7873
public T read(Kryo kryo, Input input, Class<T> aClass) {
7974
FSTObjectInput reader = fastSerialConfig.getObjectInput(input);

‎src/main/java/org/numenta/nupic/network/NetworkSerializerImpl.java‎

Lines changed: 38 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,9 @@
2121
*/
2222
package org.numenta.nupic.network;
2323

24-
import java.io.ByteArrayOutputStream;
2524
import java.io.File;
2625
import java.io.FileNotFoundException;
2726
import java.io.IOException;
28-
import java.io.InputStream;
2927
import java.nio.file.Files;
3028
import java.util.Arrays;
3129
import java.util.List;
@@ -39,16 +37,16 @@
3937
import org.joda.time.format.DateTimeFormatter;
4038
import org.numenta.nupic.Persistable;
4139
import org.nustaq.serialization.FSTConfiguration;
40+
import org.nustaq.serialization.FSTObjectInput;
4241
import org.slf4j.Logger;
4342
import org.slf4j.LoggerFactory;
4443

4544
import com.esotericsoftware.kryo.Kryo;
45+
import com.esotericsoftware.kryo.KryoException;
4646
import com.esotericsoftware.kryo.Serializer;
4747
import com.esotericsoftware.kryo.io.Input;
4848
import com.esotericsoftware.kryo.io.Output;
4949

50-
import de.javakaffee.kryoserializers.KryoReflectionFactorySupport;
51-
5250
/**
5351
* <p>
5452
* Low level serializer that wraps an FST serialization scheme in both a Kryo {@link Serializer}
@@ -67,11 +65,15 @@
6765
*
6866
* @param <T> the type which will be serialized
6967
*/
70-
class NetworkSerializerImpl<T extends Persistable> extends Serializer<T> implements NetworkSerializer<T> {
68+
class NetworkSerializerImpl<T extends Persistable> extends Serializer<T> implements NetworkSerializer<T>, Persistable {
69+
private static final long serialVersionUID = 1L;
70+
7171
protected static final Logger LOGGER = LoggerFactory.getLogger(NetworkSerializerImpl.class);
7272

7373
/** Time stamped serialization file format */
74-
public static final DateTimeFormatter CHECKPOINT_TIMESTAMP_FORMAT = DateTimeFormat.forPattern(SerialConfig.CHECKPOINT_FORMAT_STRING);
74+
public static transient DateTimeFormatter CHECKPOINT_TIMESTAMP_FORMAT = DateTimeFormat.forPattern(SerialConfig.CHECKPOINT_FORMAT_STRING);
75+
private transient DateTimeFormatter checkPointFormatter = CHECKPOINT_TIMESTAMP_FORMAT;
76+
private String checkPointFormatString;
7577

7678
/** Use of Fast Serialize https://github.com/RuedigerMoeller/fast-serialization */
7779
private final FSTConfiguration fastSerialConfig = FSTConfiguration.createDefaultConfiguration();
@@ -89,9 +91,7 @@ class NetworkSerializerImpl<T extends Persistable> extends Serializer<T> impleme
8991
/** Stores the bytes of the last serialized object or null if there was a problem */
9092
private static AtomicReference<byte[]> lastBytes = new AtomicReference<byte[]>(null);
9193

92-
private DateTimeFormatter checkPointFormatter = CHECKPOINT_TIMESTAMP_FORMAT;
9394

94-
private String checkPointFormatString;
9595

9696
/**
9797
* All instances in this classloader will share the same atomic reference to the last checkpoint file name holder
@@ -134,23 +134,21 @@ class NetworkSerializerImpl<T extends Persistable> extends Serializer<T> impleme
134134
* @param klass The class of the object to be read in.
135135
* @return an instance of type &lt;T&gt;
136136
*/
137+
@SuppressWarnings("unchecked")
137138
@Override
138139
public T read(Kryo kryo, Input in, Class<T> klass) {
139-
InputStream is = in.getInputStream();
140-
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
141-
140+
FSTObjectInput reader = fastSerialConfig.getObjectInput(in);
142141
try {
143-
int nRead;
144-
byte[] data = new byte[16384];
145-
while ((nRead = is.read(data, 0, data.length)) != -1) {
146-
buffer.write(data, 0, nRead);
142+
T t = (T) reader.readObject(klass);
143+
144+
if(t instanceof Persistable) {
145+
((Persistable) t).postDeSerialize();
147146
}
148-
buffer.flush();
149-
}catch(Exception e) {
150-
throw new RuntimeException(e);
147+
return t;
148+
}
149+
catch(Exception e) {
150+
throw new KryoException(e);
151151
}
152-
153-
return deSerialize(klass, buffer.toByteArray());
154152
}
155153

156154
/**
@@ -166,11 +164,18 @@ public T read(Kryo kryo, Input in, Class<T> klass) {
166164
* @param <T> instance to serialize
167165
*/
168166
@Override
169-
public void write(Kryo kryo, Output out, T type) {
170-
byte[] bytes = serialize(type);
171-
out.write(bytes);
172-
173-
lastBytes.set(bytes);
167+
public void write(Kryo kryo, Output out, T t) {
168+
try {
169+
if(t instanceof Persistable) {
170+
((Persistable) t).preSerialize();
171+
}
172+
173+
byte[] bytes = fastSerialConfig.asByteArray(t);
174+
out.write(bytes);
175+
}
176+
catch(Exception e) {
177+
throw new KryoException(e);
178+
}
174179
}
175180

176181
/**
@@ -547,22 +552,14 @@ File testFileExists(String fileName) throws IOException, FileNotFoundException {
547552
* @return an instance of Kryo
548553
*/
549554
private Kryo createKryo() {
550-
return new KryoReflectionFactorySupport() {
551-
private NetworkSerializer<?> ser;
552-
private SerialConfig conf;
553-
554-
{
555-
conf = new SerialConfig(
556-
config.getFileName(), Scheme.FST,
557-
config.getRegistry(), config.getOpenOptions());
558-
ser = Network.serializer(conf, true);
559-
}
560-
561-
@SuppressWarnings("rawtypes")
562-
@Override public Serializer<?> getDefaultSerializer(final Class clazz) {
563-
return (Serializer<?>)ser;
564-
}
565-
};
555+
Kryo.DefaultInstantiatorStrategy initStrategy = new Kryo.DefaultInstantiatorStrategy();
556+
557+
// use Objenesis to create classes without calling the constructor (Flink's technique)
558+
//initStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
559+
560+
Kryo kryo = new Kryo();
561+
kryo.setInstantiatorStrategy(initStrategy);
562+
return kryo;
566563
}
567564

568565
/* (non-Javadoc)

‎src/main/java/org/numenta/nupic/network/SerialConfig.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import org.numenta.nupic.model.Cell;
3636
import org.numenta.nupic.model.Column;
3737
import org.numenta.nupic.model.DistalDendrite;
38+
import org.numenta.nupic.model.Pool;
3839
import org.numenta.nupic.model.ProximalDendrite;
3940
import org.numenta.nupic.model.Segment;
4041
import org.numenta.nupic.model.Synapse;
@@ -78,11 +79,11 @@
7879
public class SerialConfig {
7980
protected static final Logger LOGGER = LoggerFactory.getLogger(NetworkSerializerImpl.class);
8081

81-
public static final Class[] DEFAULT_REGISTERED_TYPES = new Class[] {
82+
public static final Class<?>[] DEFAULT_REGISTERED_TYPES = new Class[] {
8283
Region.class, Layer.class, Cell.class, Column.class, Synapse.class,
83-
ProximalDendrite.class, DistalDendrite.class, Segment.class, Inference.class,
84-
ManualInput.class, BitHistory.class, Tuple.class, NamedTuple.class, Parameters.class,
85-
ComputeCycle.class, Classification.class, FieldMetaType.class, Persistable.class
84+
ProximalDendrite.class, DistalDendrite.class, Segment.class, Inference.class,
85+
ManualInput.class, BitHistory.class, Tuple.class, NamedTuple.class, Parameters.class,
86+
ComputeCycle.class, Classification.class, FieldMetaType.class, Pool.class, Persistable.class
8687
};
8788

8889
/** The default format for the timestamp portion of the checkpoint file name */

0 commit comments

Comments
 (0)