홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Yuta HIGUCHI
c7052013 1 parent fc9a4ff7

Fix for Kryo related issue crossing OSGi bundle boundaries. 

Change-Id: I121dfe360de14a5b4760e5d2fd8cb2db93e0be63
Inline Side-by-side
Showing 8 changed files with 171 additions and 92 deletions
...@@ -7,9 +7,6 @@ import java.net.URI; ...@@ -7,9 +7,6 @@ import java.net.URI;
7 */ 7 */
8 public final class DeviceId extends ElementId { 8 public final class DeviceId extends ElementId {
9 9
10 - // Default constructor for serialization
11 - protected DeviceId() {}
12 -
13 // Public construction is prohibited 10 // Public construction is prohibited
14 private DeviceId(URI uri) { 11 private DeviceId(URI uri) {
15 super(uri); 12 super(uri);
......
...@@ -12,12 +12,6 @@ public class ProviderId { ...@@ -12,12 +12,6 @@ public class ProviderId {
12 private final String scheme; 12 private final String scheme;
13 private final String id; 13 private final String id;
14 14
15 - // Default constructor for serialization
16 - protected ProviderId() {
17 - scheme = null;
18 - id = null;
19 - }
20 -
21 /** 15 /**
22 * Creates a new provider identifier from the specified string. 16 * Creates a new provider identifier from the specified string.
23 * The providers are expected to follow the reverse DNS convention, e.g. 17 * The providers are expected to follow the reverse DNS convention, e.g.
...@@ -40,6 +34,15 @@ public class ProviderId { ...@@ -40,6 +34,15 @@ public class ProviderId {
40 return scheme; 34 return scheme;
41 } 35 }
42 36
37 + /**
38 + * Returns the device URI scheme specific id portion.
39 + *
40 + * @return id
41 + */
42 + public String id() {
43 + return id;
44 + }
45 +
43 @Override 46 @Override
44 public int hashCode() { 47 public int hashCode() {
45 return Objects.hash(scheme, id); 48 return Objects.hash(scheme, id);
......
...@@ -112,7 +112,7 @@ public class DistributedDeviceManagerTest { ...@@ -112,7 +112,7 @@ public class DistributedDeviceManagerTest {
112 mgr.deactivate(); 112 mgr.deactivate();
113 113
114 dstore.deactivate(); 114 dstore.deactivate();
115 - dstore.theInstance.shutdown(); 115 + ((TestDistributedDeviceStore) dstore).shutdownHz();
116 } 116 }
117 117
118 private void connectDevice(DeviceId deviceId, String swVersion) { 118 private void connectDevice(DeviceId deviceId, String swVersion) {
...@@ -290,5 +290,12 @@ public class DistributedDeviceManagerTest { ...@@ -290,5 +290,12 @@ public class DistributedDeviceManagerTest {
290 } 290 }
291 }; 291 };
292 } 292 }
293 +
294 + /**
295 + * Shutdowns the hazelcast instance.
296 + */
297 + public void shutdownHz() {
298 + theInstance.shutdown();
299 + }
293 } 300 }
294 } 301 }
......
...@@ -16,6 +16,9 @@ import com.esotericsoftware.kryo.serializers.CollectionSerializer; ...@@ -16,6 +16,9 @@ import com.esotericsoftware.kryo.serializers.CollectionSerializer;
16 import com.google.common.collect.ImmutableSet; 16 import com.google.common.collect.ImmutableSet;
17 17
18 // TODO move to util, etc. 18 // TODO move to util, etc.
19 +/**
20 + * Kryo Serializer for {@link DefaultPort}.
21 + */
19 public final class DefaultPortSerializer extends 22 public final class DefaultPortSerializer extends
20 Serializer<DefaultPort> { 23 Serializer<DefaultPort> {
21 24
...@@ -23,6 +26,9 @@ public final class DefaultPortSerializer extends ...@@ -23,6 +26,9 @@ public final class DefaultPortSerializer extends
23 = new CollectionSerializer(IpPrefix.class, 26 = new CollectionSerializer(IpPrefix.class,
24 new IpPrefixSerializer(), false); 27 new IpPrefixSerializer(), false);
25 28
29 + /**
30 + * Default constructor.
31 + */
26 public DefaultPortSerializer() { 32 public DefaultPortSerializer() {
27 // non-null, immutable 33 // non-null, immutable
28 super(false, true); 34 super(false, true);
......
...@@ -44,6 +44,10 @@ import org.onlab.onos.store.StoreService; ...@@ -44,6 +44,10 @@ import org.onlab.onos.store.StoreService;
44 import org.onlab.util.KryoPool; 44 import org.onlab.util.KryoPool;
45 import org.slf4j.Logger; 45 import org.slf4j.Logger;
46 46
47 +import com.esotericsoftware.kryo.Kryo;
48 +import com.esotericsoftware.kryo.Serializer;
49 +import com.esotericsoftware.kryo.io.Input;
50 +import com.esotericsoftware.kryo.io.Output;
47 import com.google.common.base.Optional; 51 import com.google.common.base.Optional;
48 import com.google.common.cache.CacheBuilder; 52 import com.google.common.cache.CacheBuilder;
49 import com.google.common.cache.CacheLoader; 53 import com.google.common.cache.CacheLoader;
...@@ -68,103 +72,26 @@ import de.javakaffee.kryoserializers.URISerializer; ...@@ -68,103 +72,26 @@ import de.javakaffee.kryoserializers.URISerializer;
68 @Service 72 @Service
69 public class DistributedDeviceStore implements DeviceStore { 73 public class DistributedDeviceStore implements DeviceStore {
70 74
71 - /**
72 - * An IMap EntryListener, which reflects each remote event to cache.
73 - *
74 - * @param <K> IMap key type after deserialization
75 - * @param <V> IMap value type after deserialization
76 - */
77 - public static final class RemoteEventHandler<K, V> extends
78 - EntryAdapter<byte[], byte[]> {
79 -
80 - private LoadingCache<K, Optional<V>> cache;
81 -
82 - /**
83 - * Constructor.
84 - *
85 - * @param cache cache to update
86 - */
87 - public RemoteEventHandler(
88 - LoadingCache<K, Optional<V>> cache) {
89 - this.cache = checkNotNull(cache);
90 - }
91 -
92 - @Override
93 - public void mapCleared(MapEvent event) {
94 - cache.invalidateAll();
95 - }
96 -
97 - @Override
98 - public void entryUpdated(EntryEvent<byte[], byte[]> event) {
99 - cache.put(POOL.<K>deserialize(event.getKey()),
100 - Optional.of(POOL.<V>deserialize(
101 - event.getValue())));
102 - }
103 -
104 - @Override
105 - public void entryRemoved(EntryEvent<byte[], byte[]> event) {
106 - cache.invalidate(POOL.<DeviceId>deserialize(event.getKey()));
107 - }
108 -
109 - @Override
110 - public void entryAdded(EntryEvent<byte[], byte[]> event) {
111 - entryUpdated(event);
112 - }
113 - }
114 -
115 - /**
116 - * CacheLoader to wrap Map value with Optional,
117 - * to handle negative hit on underlying IMap.
118 - *
119 - * @param <K> IMap key type after deserialization
120 - * @param <V> IMap value type after deserialization
121 - */
122 - public static final class OptionalCacheLoader<K, V> extends
123 - CacheLoader<K, Optional<V>> {
124 -
125 - private IMap<byte[], byte[]> rawMap;
126 -
127 - /**
128 - * Constructor.
129 - *
130 - * @param rawMap underlying IMap
131 - */
132 - public OptionalCacheLoader(IMap<byte[], byte[]> rawMap) {
133 - this.rawMap = checkNotNull(rawMap);
134 - }
135 -
136 - @Override
137 - public Optional<V> load(K key) throws Exception {
138 - byte[] keyBytes = serialize(key);
139 - byte[] valBytes = rawMap.get(keyBytes);
140 - if (valBytes == null) {
141 - return Optional.absent();
142 - }
143 - V dev = deserialize(valBytes);
144 - return Optional.of(dev);
145 - }
146 - }
147 -
148 private final Logger log = getLogger(getClass()); 75 private final Logger log = getLogger(getClass());
149 76
150 public static final String DEVICE_NOT_FOUND = "Device with ID %s not found"; 77 public static final String DEVICE_NOT_FOUND = "Device with ID %s not found";
151 78
152 // FIXME Slice out types used in common to separate pool/namespace. 79 // FIXME Slice out types used in common to separate pool/namespace.
153 private static final KryoPool POOL = KryoPool.newBuilder() 80 private static final KryoPool POOL = KryoPool.newBuilder()
154 - .register(URI.class, new URISerializer())
155 .register( 81 .register(
156 ArrayList.class, 82 ArrayList.class,
83 + HashMap.class,
157 84
158 - ProviderId.class,
159 Device.Type.class, 85 Device.Type.class,
160 86
161 - DeviceId.class,
162 DefaultDevice.class, 87 DefaultDevice.class,
163 MastershipRole.class, 88 MastershipRole.class,
164 - HashMap.class,
165 Port.class, 89 Port.class,
166 Element.class 90 Element.class
167 ) 91 )
92 + .register(URI.class, new URISerializer())
93 + .register(ProviderId.class, new ProviderIdSerializer())
94 + .register(DeviceId.class, new DeviceIdSerializer())
168 .register(PortNumber.class, new PortNumberSerializer()) 95 .register(PortNumber.class, new PortNumberSerializer())
169 .register(DefaultPort.class, new DefaultPortSerializer()) 96 .register(DefaultPort.class, new DefaultPortSerializer())
170 .build() 97 .build()
...@@ -190,7 +117,7 @@ public class DistributedDeviceStore implements DeviceStore { ...@@ -190,7 +117,7 @@ public class DistributedDeviceStore implements DeviceStore {
190 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 117 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
191 protected StoreService storeService; 118 protected StoreService storeService;
192 119
193 - /*protected*/public HazelcastInstance theInstance; 120 + protected HazelcastInstance theInstance;
194 121
195 122
196 @Activate 123 @Activate
...@@ -517,4 +444,94 @@ public class DistributedDeviceStore implements DeviceStore { ...@@ -517,4 +444,94 @@ public class DistributedDeviceStore implements DeviceStore {
517 return POOL.deserialize(bytes); 444 return POOL.deserialize(bytes);
518 } 445 }
519 446
447 + public static final class DeviceIdSerializer extends Serializer<DeviceId> {
448 +
449 + @Override
450 + public void write(Kryo kryo, Output output, DeviceId object) {
451 + kryo.writeObject(output, object.uri());
452 + }
453 +
454 + @Override
455 + public DeviceId read(Kryo kryo, Input input, Class<DeviceId> type) {
456 + final URI uri = kryo.readObject(input, URI.class);
457 + return DeviceId.deviceId(uri);
458 + }
459 + }
460 +
461 + /**
462 + * An IMap EntryListener, which reflects each remote event to cache.
463 + *
464 + * @param <K> IMap key type after deserialization
465 + * @param <V> IMap value type after deserialization
466 + */
467 + public static final class RemoteEventHandler<K, V> extends
468 + EntryAdapter<byte[], byte[]> {
469 +
470 + private LoadingCache<K, Optional<V>> cache;
471 +
472 + /**
473 + * Constructor.
474 + *
475 + * @param cache cache to update
476 + */
477 + public RemoteEventHandler(
478 + LoadingCache<K, Optional<V>> cache) {
479 + this.cache = checkNotNull(cache);
480 + }
481 +
482 + @Override
483 + public void mapCleared(MapEvent event) {
484 + cache.invalidateAll();
485 + }
486 +
487 + @Override
488 + public void entryUpdated(EntryEvent<byte[], byte[]> event) {
489 + cache.put(POOL.<K>deserialize(event.getKey()),
490 + Optional.of(POOL.<V>deserialize(
491 + event.getValue())));
492 + }
493 +
494 + @Override
495 + public void entryRemoved(EntryEvent<byte[], byte[]> event) {
496 + cache.invalidate(POOL.<DeviceId>deserialize(event.getKey()));
497 + }
498 +
499 + @Override
500 + public void entryAdded(EntryEvent<byte[], byte[]> event) {
501 + entryUpdated(event);
502 + }
503 + }
504 +
505 + /**
506 + * CacheLoader to wrap Map value with Optional,
507 + * to handle negative hit on underlying IMap.
508 + *
509 + * @param <K> IMap key type after deserialization
510 + * @param <V> IMap value type after deserialization
511 + */
512 + public static final class OptionalCacheLoader<K, V> extends
513 + CacheLoader<K, Optional<V>> {
514 +
515 + private IMap<byte[], byte[]> rawMap;
516 +
517 + /**
518 + * Constructor.
519 + *
520 + * @param rawMap underlying IMap
521 + */
522 + public OptionalCacheLoader(IMap<byte[], byte[]> rawMap) {
523 + this.rawMap = checkNotNull(rawMap);
524 + }
525 +
526 + @Override
527 + public Optional<V> load(K key) throws Exception {
528 + byte[] keyBytes = serialize(key);
529 + byte[] valBytes = rawMap.get(keyBytes);
530 + if (valBytes == null) {
531 + return Optional.absent();
532 + }
533 + V dev = deserialize(valBytes);
534 + return Optional.of(dev);
535 + }
536 + }
520 } 537 }
......
...@@ -8,8 +8,14 @@ import com.esotericsoftware.kryo.io.Input; ...@@ -8,8 +8,14 @@ import com.esotericsoftware.kryo.io.Input;
8 import com.esotericsoftware.kryo.io.Output; 8 import com.esotericsoftware.kryo.io.Output;
9 9
10 // TODO move to util, etc. 10 // TODO move to util, etc.
11 +/**
12 + * Kryo Serializer for {@link IpPrefix}.
13 + */
11 public final class IpPrefixSerializer extends Serializer<IpPrefix> { 14 public final class IpPrefixSerializer extends Serializer<IpPrefix> {
12 15
16 + /**
17 + * Default constructor.
18 + */
13 public IpPrefixSerializer() { 19 public IpPrefixSerializer() {
14 // non-null, immutable 20 // non-null, immutable
15 super(false, true); 21 super(false, true);
......
...@@ -8,9 +8,15 @@ import com.esotericsoftware.kryo.io.Input; ...@@ -8,9 +8,15 @@ import com.esotericsoftware.kryo.io.Input;
8 import com.esotericsoftware.kryo.io.Output; 8 import com.esotericsoftware.kryo.io.Output;
9 9
10 // TODO move to util, etc. 10 // TODO move to util, etc.
11 +/**
12 + * Serializer for {@link PortNumber}.
13 + */
11 public final class PortNumberSerializer extends 14 public final class PortNumberSerializer extends
12 Serializer<PortNumber> { 15 Serializer<PortNumber> {
13 16
17 + /**
18 + * Default constructor.
19 + */
14 public PortNumberSerializer() { 20 public PortNumberSerializer() {
15 // non-null, immutable 21 // non-null, immutable
16 super(false, true); 22 super(false, true);
......
1 +package org.onlab.onos.store.device.impl;
2 +
3 +import org.onlab.onos.net.provider.ProviderId;
4 +
5 +import com.esotericsoftware.kryo.Kryo;
6 +import com.esotericsoftware.kryo.Serializer;
7 +import com.esotericsoftware.kryo.io.Input;
8 +import com.esotericsoftware.kryo.io.Output;
9 +
10 +//TODO move to util, etc.
11 +/**
12 + * Serializer for {@link ProviderId}.
13 + */
14 +public class ProviderIdSerializer extends Serializer<ProviderId> {
15 +
16 + /**
17 + * Default constructor.
18 + */
19 + public ProviderIdSerializer() {
20 + // non-null, immutable
21 + super(false, true);
22 + }
23 +
24 + @Override
25 + public void write(Kryo kryo, Output output, ProviderId object) {
26 + output.writeString(object.scheme());
27 + output.writeString(object.id());
28 + }
29 +
30 + @Override
31 + public ProviderId read(Kryo kryo, Input input, Class<ProviderId> type) {
32 + String scheme = input.readString();
33 + String id = input.readString();
34 + return new ProviderId(scheme, id);
35 + }
36 +
37 +}