Yuta HIGUCHI

moving out OptionalCacheLoader

Change-Id: If929ed119df1a0282e311188a00776e971f78991
/**
* Implementation of a distributed cluster node store using Hazelcast.
*/
package org.onlab.onos.store.cluster.impl;
\ No newline at end of file
package org.onlab.onos.store.cluster.impl;
......
......@@ -2,7 +2,6 @@ package org.onlab.onos.store.device.impl;
import com.google.common.base.Optional;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
......@@ -13,6 +12,7 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.ISet;
import com.hazelcast.core.MapEvent;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
......@@ -33,6 +33,7 @@ import org.onlab.onos.net.device.PortDescription;
import org.onlab.onos.net.provider.ProviderId;
import org.onlab.onos.store.StoreService;
import org.onlab.onos.store.impl.AbsentInvalidatingLoadingCache;
import org.onlab.onos.store.impl.OptionalCacheLoader;
import org.slf4j.Logger;
import java.util.ArrayList;
......@@ -78,7 +79,6 @@ public class DistributedDeviceStore implements DeviceStore {
private IMap<byte[], byte[]> rawDevicePorts;
private LoadingCache<DeviceId, Optional<Map<PortNumber, Port>>> devicePorts;
// FIXME change to protected once we remove DistributedDeviceManagerTest.
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected StoreService storeService;
......@@ -95,30 +95,36 @@ public class DistributedDeviceStore implements DeviceStore {
// TODO decide on Map name scheme to avoid collision
rawDevices = theInstance.getMap("devices");
final OptionalCacheLoader<DeviceId, DefaultDevice> deviceLoader
= new OptionalCacheLoader<>(storeService, rawDevices);
devices = new AbsentInvalidatingLoadingCache<>(
CacheBuilder.newBuilder()
.build(new OptionalCacheLoader<DeviceId, DefaultDevice>(rawDevices)));
.build(deviceLoader));
// refresh/populate cache based on notification from other instance
rawDevices.addEntryListener(
new RemoteEventHandler<>(devices),
includeValue);
rawRoles = theInstance.getMap("roles");
final OptionalCacheLoader<DeviceId, MastershipRole> rolesLoader
= new OptionalCacheLoader<>(storeService, rawRoles);
roles = new AbsentInvalidatingLoadingCache<>(
CacheBuilder.newBuilder()
.build(new OptionalCacheLoader<DeviceId, MastershipRole>(rawRoles)));
.build(rolesLoader));
// refresh/populate cache based on notification from other instance
rawRoles.addEntryListener(
new RemoteEventHandler<>(roles),
includeValue);
// TODO cache avai
// TODO cache availableDevices
availableDevices = theInstance.getSet("availableDevices");
rawDevicePorts = theInstance.getMap("devicePorts");
final OptionalCacheLoader<DeviceId, Map<PortNumber, Port>> devicePortLoader
= new OptionalCacheLoader<>(storeService, rawDevicePorts);
devicePorts = new AbsentInvalidatingLoadingCache<>(
CacheBuilder.newBuilder()
.build(new OptionalCacheLoader<DeviceId, Map<PortNumber, Port>>(rawDevicePorts)));
.build(devicePortLoader));
// refresh/populate cache based on notification from other instance
rawDevicePorts.addEntryListener(
new RemoteEventHandler<>(devicePorts),
......@@ -439,7 +445,7 @@ public class DistributedDeviceStore implements DeviceStore {
@Override
public void entryRemoved(EntryEvent<byte[], byte[]> event) {
cache.invalidate(storeService.<DeviceId>deserialize(event.getKey()));
cache.invalidate(storeService.<K>deserialize(event.getKey()));
}
@Override
......@@ -447,37 +453,4 @@ public class DistributedDeviceStore implements DeviceStore {
entryUpdated(event);
}
}
/**
* CacheLoader to wrap Map value with Optional,
* to handle negative hit on underlying IMap.
*
* @param <K> IMap key type after deserialization
* @param <V> IMap value type after deserialization
*/
public final class OptionalCacheLoader<K, V> extends
CacheLoader<K, Optional<V>> {
private IMap<byte[], byte[]> rawMap;
/**
* Constructor.
*
* @param rawMap underlying IMap
*/
public OptionalCacheLoader(IMap<byte[], byte[]> rawMap) {
this.rawMap = checkNotNull(rawMap);
}
@Override
public Optional<V> load(K key) throws Exception {
byte[] keyBytes = storeService.serialize(key);
byte[] valBytes = rawMap.get(keyBytes);
if (valBytes == null) {
return Optional.absent();
}
V dev = deserialize(valBytes);
return Optional.of(dev);
}
}
}
......
......@@ -7,9 +7,24 @@ import com.google.common.base.Optional;
import com.google.common.cache.ForwardingLoadingCache.SimpleForwardingLoadingCache;
import com.google.common.cache.LoadingCache;
/**
* Wrapper around LoadingCache to handle negative hit scenario.
* <p>
* When the LoadingCache returned Absent,
* this implementation will invalidate the entry immediately to avoid
* caching negative hits.
*
* @param <K> Cache key type
* @param <V> Cache value type. (Optional{@literal <V>})
*/
public class AbsentInvalidatingLoadingCache<K, V> extends
SimpleForwardingLoadingCache<K, Optional<V>> {
/**
* Constructor.
*
* @param delegate actual {@link LoadingCache} to delegate loading.
*/
public AbsentInvalidatingLoadingCache(LoadingCache<K, Optional<V>> delegate) {
super(delegate);
}
......
package org.onlab.onos.store.impl;
import static com.google.common.base.Preconditions.checkNotNull;
import org.onlab.onos.store.StoreService;
import com.google.common.base.Optional;
import com.google.common.cache.CacheLoader;
import com.hazelcast.core.IMap;
/**
* CacheLoader to wrap Map value with Optional,
* to handle negative hit on underlying IMap.
*
* @param <K> IMap key type after deserialization
* @param <V> IMap value type after deserialization
*/
public final class OptionalCacheLoader<K, V> extends
CacheLoader<K, Optional<V>> {
private final StoreService storeService;
private IMap<byte[], byte[]> rawMap;
/**
* Constructor.
*
* @param storeService to use for serialization
* @param rawMap underlying IMap
*/
public OptionalCacheLoader(StoreService storeService, IMap<byte[], byte[]> rawMap) {
this.storeService = checkNotNull(storeService);
this.rawMap = checkNotNull(rawMap);
}
@Override
public Optional<V> load(K key) throws Exception {
byte[] keyBytes = storeService.serialize(key);
byte[] valBytes = rawMap.get(keyBytes);
if (valBytes == null) {
return Optional.absent();
}
V dev = storeService.deserialize(valBytes);
return Optional.of(dev);
}
}
/**
* Various Kryo serializers for use in distributed stores.
*/
package org.onlab.onos.store.serializers;
\ No newline at end of file
package org.onlab.onos.store.serializers;
......