Ayaka Koshibe

Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next

1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import static org.onlab.onos.store.cluster.messaging.MessageSubject.AE_ADVERTISEMENT;
4 +import java.util.Map;
5 +
6 +import org.onlab.onos.cluster.NodeId;
7 +import org.onlab.onos.store.Timestamp;
8 +
9 +import com.google.common.collect.ImmutableMap;
10 +
11 +/**
12 + * Anti-Entropy advertisement message.
13 + *
14 + * @param <ID> ID type
15 + */
16 +public class AntiEntropyAdvertisement<ID> extends ClusterMessage {
17 +
18 + private final NodeId sender;
19 + private final ImmutableMap<ID, Timestamp> advertisement;
20 +
21 + /**
22 + * Creates anti-entropy advertisement message.
23 + *
24 + * @param sender sender of this message
25 + * @param advertisement timestamp information of the data sender holds
26 + */
27 + public AntiEntropyAdvertisement(NodeId sender, Map<ID, Timestamp> advertisement) {
28 + super(AE_ADVERTISEMENT);
29 + this.sender = sender;
30 + this.advertisement = ImmutableMap.copyOf(advertisement);
31 + }
32 +
33 + public NodeId sender() {
34 + return sender;
35 + }
36 +
37 + public ImmutableMap<ID, Timestamp> advertisement() {
38 + return advertisement;
39 + }
40 +
41 + // Default constructor for serializer
42 + protected AntiEntropyAdvertisement() {
43 + super(AE_ADVERTISEMENT);
44 + this.sender = null;
45 + this.advertisement = null;
46 + }
47 +}
1 +package org.onlab.onos.store.cluster.messaging;
2 +
3 +import static org.onlab.onos.store.cluster.messaging.MessageSubject.AE_REPLY;
4 +
5 +import java.util.Map;
6 +import java.util.Set;
7 +
8 +import org.onlab.onos.cluster.NodeId;
9 +import org.onlab.onos.store.device.impl.VersionedValue;
10 +
11 +import com.google.common.collect.ImmutableMap;
12 +import com.google.common.collect.ImmutableSet;
13 +
14 +public class AntiEntropyReply<ID, VALUE> extends ClusterMessage {
15 +
16 + private final NodeId sender;
17 + private final ImmutableMap<ID, VersionedValue<VALUE>> suggestion;
18 + private final ImmutableSet<ID> request;
19 +
20 + /**
21 + * Creates a reply to anti-entropy message.
22 + *
23 + * @param sender sender of this message
24 + * @param suggestion collection of more recent values, sender had
25 + * @param request Collection of identifiers
26 + */
27 + public AntiEntropyReply(NodeId sender,
28 + Map<ID, VersionedValue<VALUE>> suggestion,
29 + Set<ID> request) {
30 + super(AE_REPLY);
31 + this.sender = sender;
32 + this.suggestion = ImmutableMap.copyOf(suggestion);
33 + this.request = ImmutableSet.copyOf(request);
34 + }
35 +
36 + public NodeId sender() {
37 + return sender;
38 + }
39 +
40 + public ImmutableMap<ID, VersionedValue<VALUE>> suggestion() {
41 + return suggestion;
42 + }
43 +
44 + public ImmutableSet<ID> request() {
45 + return request;
46 + }
47 +
48 + // Default constructor for serializer
49 + protected AntiEntropyReply() {
50 + super(AE_REPLY);
51 + this.sender = null;
52 + this.suggestion = null;
53 + this.request = null;
54 + }
55 +}
...@@ -15,6 +15,12 @@ public enum MessageSubject { ...@@ -15,6 +15,12 @@ public enum MessageSubject {
15 LEAVING_MEMBER, 15 LEAVING_MEMBER,
16 16
17 /** Signifies a heart-beat message. */ 17 /** Signifies a heart-beat message. */
18 - ECHO 18 + ECHO,
19 +
20 + /** Anti-Entropy advertisement message. */
21 + AE_ADVERTISEMENT,
22 +
23 + /** Anti-Entropy reply message. */
24 + AE_REPLY,
19 25
20 } 26 }
......
...@@ -42,4 +42,12 @@ public class VersionedValue<T> { ...@@ -42,4 +42,12 @@ public class VersionedValue<T> {
42 public Timestamp timestamp() { 42 public Timestamp timestamp() {
43 return timestamp; 43 return timestamp;
44 } 44 }
45 +
46 +
47 + // Default constructor for serializer
48 + protected VersionedValue() {
49 + this.entity = null;
50 + this.isUp = false;
51 + this.timestamp = null;
52 + }
45 } 53 }
......
1 +package org.onlab.onos.store.serializers;
2 +
3 +import java.util.Collections;
4 +import java.util.HashMap;
5 +import java.util.Map;
6 +
7 +import org.onlab.util.KryoPool.FamilySerializer;
8 +
9 +import com.esotericsoftware.kryo.Kryo;
10 +import com.esotericsoftware.kryo.io.Input;
11 +import com.esotericsoftware.kryo.io.Output;
12 +import com.esotericsoftware.kryo.serializers.MapSerializer;
13 +import com.google.common.collect.ImmutableMap;
14 +
15 +/**
16 +* Kryo Serializer for {@link ImmutableMap}.
17 +*/
18 +public class ImmutableMapSerializer extends FamilySerializer<ImmutableMap<?, ?>> {
19 +
20 + private final MapSerializer mapSerializer = new MapSerializer();
21 +
22 + public ImmutableMapSerializer() {
23 + // non-null, immutable
24 + super(false, true);
25 + }
26 +
27 + @Override
28 + public void write(Kryo kryo, Output output, ImmutableMap<?, ?> object) {
29 + // wrapping with unmodifiableMap proxy
30 + // to avoid Kryo from writing only the reference marker of this instance,
31 + // which will be embedded right before this method call.
32 + kryo.writeObject(output, Collections.unmodifiableMap(object), mapSerializer);
33 + }
34 +
35 + @Override
36 + public ImmutableMap<?, ?> read(Kryo kryo, Input input,
37 + Class<ImmutableMap<?, ?>> type) {
38 + Map<?, ?> map = kryo.readObject(input, HashMap.class, mapSerializer);
39 + return ImmutableMap.copyOf(map);
40 + }
41 +
42 + @Override
43 + public void registerFamilies(Kryo kryo) {
44 + kryo.register(ImmutableMap.of().getClass(), this);
45 + kryo.register(ImmutableMap.of(1, 2).getClass(), this);
46 + kryo.register(ImmutableMap.of(1, 2, 3, 4).getClass(), this);
47 + // TODO register required ImmutableMap variants
48 + }
49 +}
1 +package org.onlab.onos.store.serializers;
2 +
3 +import java.util.ArrayList;
4 +import java.util.List;
5 +
6 +import org.onlab.util.KryoPool.FamilySerializer;
7 +
8 +import com.esotericsoftware.kryo.Kryo;
9 +import com.esotericsoftware.kryo.io.Input;
10 +import com.esotericsoftware.kryo.io.Output;
11 +import com.esotericsoftware.kryo.serializers.CollectionSerializer;
12 +import com.google.common.collect.ImmutableSet;
13 +
14 +/**
15 +* Kryo Serializer for {@link ImmutableSet}.
16 +*/
17 +public class ImmutableSetSerializer extends FamilySerializer<ImmutableSet<?>> {
18 +
19 + private final CollectionSerializer serializer = new CollectionSerializer();
20 +
21 + public ImmutableSetSerializer() {
22 + // non-null, immutable
23 + super(false, true);
24 + }
25 +
26 + @Override
27 + public void write(Kryo kryo, Output output, ImmutableSet<?> object) {
28 + kryo.writeObject(output, object.asList(), serializer);
29 + }
30 +
31 + @Override
32 + public ImmutableSet<?> read(Kryo kryo, Input input,
33 + Class<ImmutableSet<?>> type) {
34 + List<?> elms = kryo.readObject(input, ArrayList.class, serializer);
35 + return ImmutableSet.copyOf(elms);
36 + }
37 +
38 + @Override
39 + public void registerFamilies(Kryo kryo) {
40 + kryo.register(ImmutableSet.of().getClass(), this);
41 + kryo.register(ImmutableSet.of(1).getClass(), this);
42 + kryo.register(ImmutableSet.of(1, 2).getClass(), this);
43 + // TODO register required ImmutableSet variants
44 + }
45 +}
1 +package org.onlab.onos.store.serializers;
2 +
3 +import static org.onlab.onos.net.DeviceId.deviceId;
4 +import static org.onlab.onos.net.PortNumber.portNumber;
5 +
6 +import java.net.URI;
7 +import java.nio.ByteBuffer;
8 +import java.util.ArrayList;
9 +import java.util.HashMap;
10 +
11 +import org.junit.After;
12 +import org.junit.Before;
13 +import org.junit.BeforeClass;
14 +import org.junit.Test;
15 +import org.onlab.onos.cluster.NodeId;
16 +import org.onlab.onos.net.ConnectPoint;
17 +import org.onlab.onos.net.DefaultDevice;
18 +import org.onlab.onos.net.DefaultLink;
19 +import org.onlab.onos.net.DefaultPort;
20 +import org.onlab.onos.net.Device;
21 +import org.onlab.onos.net.DeviceId;
22 +import org.onlab.onos.net.Link;
23 +import org.onlab.onos.net.LinkKey;
24 +import org.onlab.onos.net.PortNumber;
25 +import org.onlab.onos.net.provider.ProviderId;
26 +import org.onlab.packet.IpPrefix;
27 +import org.onlab.util.KryoPool;
28 +
29 +import com.google.common.collect.ImmutableMap;
30 +import com.google.common.collect.ImmutableSet;
31 +import com.google.common.testing.EqualsTester;
32 +
33 +import de.javakaffee.kryoserializers.URISerializer;
34 +
35 +public class KryoSerializerTests {
36 + private static final ProviderId PID = new ProviderId("of", "foo");
37 + private static final DeviceId DID1 = deviceId("of:foo");
38 + private static final DeviceId DID2 = deviceId("of:bar");
39 + private static final PortNumber P1 = portNumber(1);
40 + private static final PortNumber P2 = portNumber(2);
41 + private static final ConnectPoint CP1 = new ConnectPoint(DID1, P1);
42 + private static final ConnectPoint CP2 = new ConnectPoint(DID2, P2);
43 + private static final String MFR = "whitebox";
44 + private static final String HW = "1.1.x";
45 + private static final String SW1 = "3.8.1";
46 + private static final String SW2 = "3.9.5";
47 + private static final String SN = "43311-12345";
48 + private static final Device DEV1 = new DefaultDevice(PID, DID1, Device.Type.SWITCH, MFR, HW, SW1, SN);
49 +
50 + private static KryoPool kryos;
51 +
52 + @BeforeClass
53 + public static void setUpBeforeClass() throws Exception {
54 + kryos = KryoPool.newBuilder()
55 + .register(
56 + ArrayList.class,
57 + HashMap.class
58 + )
59 + .register(
60 + Device.Type.class,
61 + Link.Type.class
62 +
63 +// ControllerNode.State.class,
64 +// DefaultControllerNode.class,
65 +// MastershipRole.class,
66 +// Port.class,
67 +// Element.class,
68 + )
69 + .register(ConnectPoint.class, new ConnectPointSerializer())
70 + .register(DefaultLink.class, new DefaultLinkSerializer())
71 + .register(DefaultPort.class, new DefaultPortSerializer())
72 + .register(DeviceId.class, new DeviceIdSerializer())
73 + .register(ImmutableMap.class, new ImmutableMapSerializer())
74 + .register(ImmutableSet.class, new ImmutableSetSerializer())
75 + .register(IpPrefix.class, new IpPrefixSerializer())
76 + .register(LinkKey.class, new LinkKeySerializer())
77 + .register(NodeId.class, new NodeIdSerializer())
78 + .register(PortNumber.class, new PortNumberSerializer())
79 + .register(ProviderId.class, new ProviderIdSerializer())
80 +
81 + .register(DefaultDevice.class)
82 +
83 + .register(URI.class, new URISerializer())
84 + .build();
85 + }
86 +
87 + @Before
88 + public void setUp() throws Exception {
89 + }
90 +
91 + @After
92 + public void tearDown() throws Exception {
93 + // removing Kryo instance to use fresh Kryo on each tests
94 + kryos.getKryo();
95 + }
96 +
97 + private static <T> void testSerialized(T original) {
98 + ByteBuffer buffer = ByteBuffer.allocate(1 * 1024 * 1024);
99 + kryos.serialize(original, buffer);
100 + buffer.flip();
101 + T copy = kryos.deserialize(buffer);
102 +
103 + new EqualsTester()
104 + .addEqualityGroup(original, copy)
105 + .testEquals();
106 + }
107 +
108 +
109 + @Test
110 + public final void test() {
111 + testSerialized(new ConnectPoint(DID1, P1));
112 + testSerialized(new DefaultLink(PID, CP1, CP2, Link.Type.DIRECT));
113 + testSerialized(new DefaultPort(DEV1, P1, true));
114 + testSerialized(DID1);
115 + testSerialized(ImmutableMap.of(DID1, DEV1, DID2, DEV1));
116 + testSerialized(ImmutableMap.of(DID1, DEV1));
117 + testSerialized(ImmutableMap.of());
118 + testSerialized(ImmutableSet.of(DID1, DID2));
119 + testSerialized(ImmutableSet.of(DID1));
120 + testSerialized(ImmutableSet.of());
121 + testSerialized(IpPrefix.valueOf("192.168.0.1/24"));
122 + testSerialized(new LinkKey(CP1, CP2));
123 + testSerialized(new NodeId("SomeNodeIdentifier"));
124 + testSerialized(P1);
125 + testSerialized(PID);
126 + }
127 +
128 +}
1 -unset OC1 OC2 OC3 OC4 OC5 OC6 OC7 OC8 OC9 OCN ONOS_NIC 1 +unset OC1 OC2 OC3 OC4 OC5 OC6 OC7 OC8 OC9 OCN ONOS_NIC ONOS_FEATURES
......
...@@ -239,12 +239,41 @@ public final class KryoPool { ...@@ -239,12 +239,41 @@ public final class KryoPool {
239 Kryo kryo = new Kryo(); 239 Kryo kryo = new Kryo();
240 kryo.setRegistrationRequired(registrationRequired); 240 kryo.setRegistrationRequired(registrationRequired);
241 for (Pair<Class<?>, Serializer<?>> registry : registeredTypes) { 241 for (Pair<Class<?>, Serializer<?>> registry : registeredTypes) {
242 - if (registry.getRight() == null) { 242 + final Serializer<?> serializer = registry.getRight();
243 + if (serializer == null) {
243 kryo.register(registry.getLeft()); 244 kryo.register(registry.getLeft());
244 } else { 245 } else {
245 - kryo.register(registry.getLeft(), registry.getRight()); 246 + kryo.register(registry.getLeft(), serializer);
247 + if (serializer instanceof FamilySerializer) {
248 + FamilySerializer<?> fser = (FamilySerializer<?>) serializer;
249 + fser.registerFamilies(kryo);
250 + }
246 } 251 }
247 } 252 }
248 return kryo; 253 return kryo;
249 } 254 }
255 +
256 + /**
257 + * Serializer implementation, which required registration of family of Classes.
258 + * @param <T> base type of this serializer.
259 + */
260 + public abstract static class FamilySerializer<T> extends Serializer<T> {
261 +
262 +
263 + public FamilySerializer(boolean acceptsNull) {
264 + super(acceptsNull);
265 + }
266 +
267 + public FamilySerializer(boolean acceptsNull, boolean immutable) {
268 + super(acceptsNull, immutable);
269 + }
270 +
271 + /**
272 + * Registers other classes this Serializer supports.
273 + *
274 + * @param kryo instance to register classes to
275 + */
276 + public void registerFamilies(Kryo kryo) {
277 + }
278 + }
250 } 279 }
......