mirrors/onos
1
0
Fork 0
mirror of https://github.com/opennetworkinglab/onos.git synced 2026-05-05 20:26:16 +02:00
Code Issues Projects Releases Wiki Activity

Dropping DatabaseManager and related code. Goodbye!

Browse Source 
Change-Id: I5d90d62678402234462dad8be455903de481da21
This commit is contained in:
Madan Jampani 2016-04-04 20:28:08 -07:00 committed by Gerrit Code Review
parent e04626fc2d
commit 38f01690db
34 changed files with 3 additions and 4924 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AsyncCachingConsistentMap.java
View File

@ -1,81 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.concurrent.CompletableFuture;
import org.onosproject.core.ApplicationId;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.Versioned;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
/**
* Extension of {@link DefaultAsyncConsistentMap} that provides a weaker read consistency
* guarantee in return for better read performance.
* <p>
* For read/write operations that are local to a node this map implementation provides
* guarantees similar to a ConsistentMap. However for read/write operations executed
* across multiple nodes this implementation only provides eventual consistency.
*
* @param <K> key type
* @param <V> value type
*/
public class AsyncCachingConsistentMap<K, V> extends DefaultAsyncConsistentMap<K, V> {
private final LoadingCache<K, CompletableFuture<Versioned<V>>> cache =
CacheBuilder.newBuilder()
.maximumSize(10000) // TODO: make configurable
.build(new CacheLoader<K, CompletableFuture<Versioned<V>>>() {
@Override
public CompletableFuture<Versioned<V>> load(K key)
throws Exception {
return AsyncCachingConsistentMap.super.get(key);
}
});
public AsyncCachingConsistentMap(String name,
ApplicationId applicationId,
Database database,
Serializer serializer,
boolean readOnly,
boolean purgeOnUninstall,
boolean meteringEnabled) {
super(name, applicationId, database, serializer, readOnly, purgeOnUninstall, meteringEnabled);
addListener(event -> cache.invalidate(event.key()));
}
@Override
public CompletableFuture<Versioned<V>> get(K key) {
CompletableFuture<Versioned<V>> cachedValue = cache.getIfPresent(key);
if (cachedValue != null) {
if (cachedValue.isCompletedExceptionally()) {
cache.invalidate(key);
} else {
return cachedValue;
}
}
return cache.getUnchecked(key);
}
@Override
protected void beforeUpdate(K key) {
super.beforeUpdate(key);
cache.invalidate(key);
}
}

61
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CommitResponse.java
View File

@ -1,61 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static com.google.common.base.MoreObjects.toStringHelper;
import java.util.Collections;
import java.util.List;
import com.google.common.collect.ImmutableList;
/**
* Result of a Transaction commit operation.
*/
public final class CommitResponse {
private boolean success;
private List<UpdateResult<String, byte[]>> updates;
public static CommitResponse success(List<UpdateResult<String, byte[]>> updates) {
return new CommitResponse(true, updates);
}
public static CommitResponse failure() {
return new CommitResponse(false, Collections.emptyList());
}
private CommitResponse(boolean success, List<UpdateResult<String, byte[]>> updates) {
this.success = success;
this.updates = ImmutableList.copyOf(updates);
}
public boolean success() {
return success;
}
public List<UpdateResult<String, byte[]>> updates() {
return updates;
}
@Override
public String toString() {
return toStringHelper(this)
.add("success", success)
.add("udpates", updates)
.toString();
}
}

134
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatCommunicationProtocol.java
View File

@ -1,134 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import org.onlab.util.Tools;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
import org.onosproject.store.cluster.messaging.MessageSubject;
import net.kuujo.copycat.protocol.AbstractProtocol;
import net.kuujo.copycat.protocol.ProtocolClient;
import net.kuujo.copycat.protocol.ProtocolHandler;
import net.kuujo.copycat.protocol.ProtocolServer;
import net.kuujo.copycat.util.Configurable;
/**
* Protocol for Copycat communication that employs
* {@code ClusterCommunicationService}.
*/
public class CopycatCommunicationProtocol extends AbstractProtocol {
private static final MessageSubject COPYCAT_MESSAGE_SUBJECT =
new MessageSubject("onos-copycat-message");
protected ClusterService clusterService;
protected ClusterCommunicationService clusterCommunicator;
public CopycatCommunicationProtocol(ClusterService clusterService,
ClusterCommunicationService clusterCommunicator) {
this.clusterService = clusterService;
this.clusterCommunicator = clusterCommunicator;
}
@Override
public Configurable copy() {
return this;
}
@Override
public ProtocolClient createClient(URI uri) {
NodeId nodeId = uriToNodeId(uri);
if (nodeId == null) {
throw new IllegalStateException("Unknown peer " + uri);
}
return new Client(nodeId);
}
@Override
public ProtocolServer createServer(URI uri) {
return new Server();
}
private class Server implements ProtocolServer {
@Override
public void handler(ProtocolHandler handler) {
if (handler == null) {
clusterCommunicator.removeSubscriber(COPYCAT_MESSAGE_SUBJECT);
} else {
clusterCommunicator.addSubscriber(COPYCAT_MESSAGE_SUBJECT,
ByteBuffer::wrap,
handler,
Tools::byteBuffertoArray);
// FIXME: Tools::byteBuffertoArray involves a array copy.
}
}
@Override
public CompletableFuture<Void> listen() {
return CompletableFuture.completedFuture(null);
}
@Override
public CompletableFuture<Void> close() {
clusterCommunicator.removeSubscriber(COPYCAT_MESSAGE_SUBJECT);
return CompletableFuture.completedFuture(null);
}
}
private class Client implements ProtocolClient {
private final NodeId peer;
public Client(NodeId peer) {
this.peer = peer;
}
@Override
public CompletableFuture<ByteBuffer> write(ByteBuffer request) {
return clusterCommunicator.sendAndReceive(request,
COPYCAT_MESSAGE_SUBJECT,
Tools::byteBuffertoArray,
ByteBuffer::wrap,
peer);
}
@Override
public CompletableFuture<Void> connect() {
return CompletableFuture.completedFuture(null);
}
@Override
public CompletableFuture<Void> close() {
return CompletableFuture.completedFuture(null);
}
}
private NodeId uriToNodeId(URI uri) {
return clusterService.getNodes()
.stream()
.filter(node -> uri.getHost().equals(node.ip().toString()))
.map(ControllerNode::id)
.findAny()
.orElse(null);
}
}

106
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Database.java
View File

@ -1,106 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.function.Consumer;
import net.kuujo.copycat.cluster.ClusterConfig;
import net.kuujo.copycat.cluster.internal.coordinator.ClusterCoordinator;
import net.kuujo.copycat.cluster.internal.coordinator.CoordinatorConfig;
import net.kuujo.copycat.cluster.internal.coordinator.DefaultClusterCoordinator;
import net.kuujo.copycat.resource.Resource;
/**
* Database.
*/
public interface Database extends DatabaseProxy<String, byte[]>, Resource<Database> {
/**
* Creates a new database with the default cluster configuration.<p>
*
* The database will be constructed with the default cluster configuration. The default cluster configuration
* searches for two resources on the classpath - {@code cluster} and {cluster-defaults} - in that order. Configuration
* options specified in {@code cluster.conf} will override those in {cluster-defaults.conf}.<p>
*
* Additionally, the database will be constructed with an database configuration that searches the classpath for
* three configuration files - {@code name}, {@code database}, {@code database-defaults}, {@code resource}, and
* {@code resource-defaults} - in that order. The first resource is a configuration resource with the same name
* as the map resource. If the resource is namespaced - e.g. `databases.my-database.conf` - then resource
* configurations will be loaded according to namespaces as well; for example, `databases.conf`.
*
* @param name The database name.
* @return The database.
*/
static Database create(String name) {
return create(name, new ClusterConfig(), new DatabaseConfig());
}
/**
* Creates a new database.<p>
*
* The database will be constructed with an database configuration that searches the classpath for
* three configuration files - {@code name}, {@code database}, {@code database-defaults}, {@code resource}, and
* {@code resource-defaults} - in that order. The first resource is a configuration resource with the same name
* as the database resource. If the resource is namespaced - e.g. `databases.my-database.conf` - then resource
* configurations will be loaded according to namespaces as well; for example, `databases.conf`.
*
* @param name The database name.
* @param cluster The cluster configuration.
* @return The database.
*/
static Database create(String name, ClusterConfig cluster) {
return create(name, cluster, new DatabaseConfig());
}
/**
* Creates a new database.
*
* @param name The database name.
* @param cluster The cluster configuration.
* @param config The database configuration.
* @return The database.
*/
static Database create(String name, ClusterConfig cluster, DatabaseConfig config) {
ClusterCoordinator coordinator =
new DefaultClusterCoordinator(new CoordinatorConfig().withName(name).withClusterConfig(cluster));
return coordinator.<Database>getResource(name, config.resolve(cluster))
.addStartupTask(() -> coordinator.open().thenApply(v -> null))
.addShutdownTask(coordinator::close);
}
/**
* Tells whether the database supports change notifications.
* @return true if notifications are supported; false otherwise
*/
default boolean hasChangeNotificationSupport() {
return true;
}
/**
* Registers a new consumer of StateMachineUpdates.
* @param consumer consumer to register
*/
void registerConsumer(Consumer<StateMachineUpdate> consumer);
/**
* Unregisters a consumer of StateMachineUpdates.
* @param consumer consumer to unregister
*/
void unregisterConsumer(Consumer<StateMachineUpdate> consumer);
}

157
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseConfig.java
View File

@ -1,157 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import com.typesafe.config.ConfigValueFactory;
import net.kuujo.copycat.cluster.ClusterConfig;
import net.kuujo.copycat.cluster.internal.coordinator.CoordinatedResourceConfig;
import net.kuujo.copycat.protocol.Consistency;
import net.kuujo.copycat.resource.ResourceConfig;
import net.kuujo.copycat.state.StateLogConfig;
import net.kuujo.copycat.util.internal.Assert;
import java.util.Map;
/**
* Database configuration.
*
*/
public class DatabaseConfig extends ResourceConfig<DatabaseConfig> {
private static final String DATABASE_CONSISTENCY = "consistency";
private static final String DEFAULT_CONFIGURATION = "database-defaults";
private static final String CONFIGURATION = "database";
private String name;
public DatabaseConfig() {
super(CONFIGURATION, DEFAULT_CONFIGURATION);
}
public DatabaseConfig(Map<String, Object> config) {
super(config, CONFIGURATION, DEFAULT_CONFIGURATION);
}
public DatabaseConfig(String resource) {
super(resource, CONFIGURATION, DEFAULT_CONFIGURATION);
}
protected DatabaseConfig(DatabaseConfig config) {
super(config);
}
@Override
public DatabaseConfig copy() {
return new DatabaseConfig(this);
}
/**
* Sets the database read consistency.
*
* @param consistency The database read consistency.
* @throws java.lang.NullPointerException If the consistency is {@code null}
*/
public void setConsistency(String consistency) {
this.config = config.withValue(DATABASE_CONSISTENCY,
ConfigValueFactory.fromAnyRef(
Consistency.parse(Assert.isNotNull(consistency, "consistency")).toString()));
}
/**
* Sets the database read consistency.
*
* @param consistency The database read consistency.
* @throws java.lang.NullPointerException If the consistency is {@code null}
*/
public void setConsistency(Consistency consistency) {
this.config = config.withValue(DATABASE_CONSISTENCY,
ConfigValueFactory.fromAnyRef(
Assert.isNotNull(consistency, "consistency").toString()));
}
/**
* Returns the database read consistency.
*
* @return The database read consistency.
*/
public Consistency getConsistency() {
return Consistency.parse(config.getString(DATABASE_CONSISTENCY));
}
/**
* Sets the database read consistency, returning the configuration for method chaining.
*
* @param consistency The database read consistency.
* @return The database configuration.
* @throws java.lang.NullPointerException If the consistency is {@code null}
*/
public DatabaseConfig withConsistency(String consistency) {
setConsistency(consistency);
return this;
}
/**
* Sets the database read consistency, returning the configuration for method chaining.
*
* @param consistency The database read consistency.
* @return The database configuration.
* @throws java.lang.NullPointerException If the consistency is {@code null}
*/
public DatabaseConfig withConsistency(Consistency consistency) {
setConsistency(consistency);
return this;
}
/**
* Returns the database name.
*
* @return The database name
*/
public String getName() {
return name;
}
/**
* Sets the database name, returning the configuration for method chaining.
*
* @param name The database name
* @return The database configuration
* @throws java.lang.NullPointerException If the name is {@code null}
*/
public DatabaseConfig withName(String name) {
setName(Assert.isNotNull(name, "name"));
return this;
}
/**
* Sets the database name.
*
* @param name The database name
* @throws java.lang.NullPointerException If the name is {@code null}
*/
public void setName(String name) {
this.name = Assert.isNotNull(name, "name");
}
@Override
public CoordinatedResourceConfig resolve(ClusterConfig cluster) {
return new StateLogConfig(toMap())
.resolve(cluster)
.withResourceType(DefaultDatabase.class);
}
}

456
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseManager.java
View File

@ -1,456 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static org.onosproject.app.ApplicationEvent.Type.APP_DEACTIVATED;
import static org.onosproject.app.ApplicationEvent.Type.APP_UNINSTALLED;
import static org.slf4j.LoggerFactory.getLogger;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import net.kuujo.copycat.CopycatConfig;
import net.kuujo.copycat.cluster.ClusterConfig;
import net.kuujo.copycat.cluster.Member;
import net.kuujo.copycat.cluster.Member.Type;
import net.kuujo.copycat.cluster.internal.coordinator.ClusterCoordinator;
import net.kuujo.copycat.cluster.internal.coordinator.DefaultClusterCoordinator;
import net.kuujo.copycat.log.BufferedLog;
import net.kuujo.copycat.log.FileLog;
import net.kuujo.copycat.log.Log;
import net.kuujo.copycat.protocol.Consistency;
import net.kuujo.copycat.protocol.Protocol;
import net.kuujo.copycat.util.concurrent.NamedThreadFactory;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.ReferencePolicy;
import org.apache.felix.scr.annotations.Service;
import org.onosproject.app.ApplicationEvent;
import org.onosproject.app.ApplicationListener;
import org.onosproject.app.ApplicationService;
import org.onosproject.cluster.ClusterMetadataService;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.cluster.PartitionId;
import org.onosproject.core.ApplicationId;
import org.onosproject.persistence.PersistenceService;
import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.serializers.KryoNamespaces;
import org.onosproject.store.service.AsyncConsistentMap;
import org.onosproject.store.service.AtomicCounterBuilder;
import org.onosproject.store.service.AtomicValueBuilder;
import org.onosproject.store.service.ConsistentMapBuilder;
import org.onosproject.store.service.ConsistentMapException;
import org.onosproject.store.service.DistributedQueueBuilder;
import org.onosproject.store.service.DistributedSetBuilder;
import org.onosproject.store.service.EventuallyConsistentMapBuilder;
import org.onosproject.store.service.LeaderElectorBuilder;
import org.onosproject.store.service.MapInfo;
import org.onosproject.store.service.PartitionInfo;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.StorageAdminService;
import org.onosproject.store.service.StorageService;
import org.onosproject.store.service.TransactionContextBuilder;
import org.slf4j.Logger;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Futures;
/**
* Database manager.
*/
@Component(immediate = true, enabled = false)
@Service
public class DatabaseManager implements StorageService, StorageAdminService {
private final Logger log = getLogger(getClass());
public static final String BASE_PARTITION_NAME = "p0";
private static final int RAFT_ELECTION_TIMEOUT_MILLIS = 3000;
private static final int DATABASE_OPERATION_TIMEOUT_MILLIS = 5000;
private ClusterCoordinator coordinator;
protected PartitionedDatabase partitionedDatabase;
protected Database inMemoryDatabase;
protected NodeId localNodeId;
private TransactionManager transactionManager;
private final Supplier<TransactionId> transactionIdGenerator =
() -> TransactionId.from(UUID.randomUUID().toString());
private ApplicationListener appListener = new InternalApplicationListener();
private final Multimap<String, DefaultAsyncConsistentMap> maps =
Multimaps.synchronizedMultimap(ArrayListMultimap.create());
private final Multimap<ApplicationId, DefaultAsyncConsistentMap> mapsByApplication =
Multimaps.synchronizedMultimap(ArrayListMultimap.create());
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClusterMetadataService clusterMetadataService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClusterService clusterService;
@Reference(cardinality = ReferenceCardinality.OPTIONAL_UNARY, policy = ReferencePolicy.DYNAMIC)
protected ApplicationService applicationService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClusterCommunicationService clusterCommunicator;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected PersistenceService persistenceService;
protected String nodeIdToUri(NodeId nodeId) {
ControllerNode node = clusterService.getNode(nodeId);
return String.format("onos://%s:%d", node.ip(), node.tcpPort());
}
protected void bindApplicationService(ApplicationService service) {
applicationService = service;
applicationService.addListener(appListener);
}
protected void unbindApplicationService(ApplicationService service) {
applicationService.removeListener(appListener);
this.applicationService = null;
}
@Activate
public void activate() {
localNodeId = clusterService.getLocalNode().id();
Map<PartitionId, Set<NodeId>> partitionMap = Maps.newHashMap();
clusterMetadataService.getClusterMetadata().getPartitions().forEach(p -> {
partitionMap.put(p.getId(), Sets.newHashSet(p.getMembers()));
});
String[] activeNodeUris = partitionMap.values()
.stream()
.reduce((s1, s2) -> Sets.union(s1, s2))
.get()
.stream()
.map(this::nodeIdToUri)
.toArray(String[]::new);
String localNodeUri = nodeIdToUri(clusterMetadataService.getLocalNode().id());
Protocol protocol = new CopycatCommunicationProtocol(clusterService, clusterCommunicator);
ClusterConfig clusterConfig = new ClusterConfig()
.withProtocol(protocol)
.withElectionTimeout(electionTimeoutMillis(activeNodeUris))
.withHeartbeatInterval(heartbeatTimeoutMillis(activeNodeUris))
.withMembers(activeNodeUris)
.withLocalMember(localNodeUri);
CopycatConfig copycatConfig = new CopycatConfig()
.withName("onos")
.withClusterConfig(clusterConfig)
.withDefaultSerializer(new DatabaseSerializer())
.withDefaultExecutor(Executors.newSingleThreadExecutor(new NamedThreadFactory("copycat-coordinator-%d")));
coordinator = new DefaultClusterCoordinator(copycatConfig.resolve());
Function<PartitionId, Log> logFunction = id -> id.asInt() == 0 ? newInMemoryLog() : newPersistentLog();
Map<PartitionId, Database> databases = Maps.transformEntries(partitionMap, (k, v) -> {
String[] replicas = v.stream().map(this::nodeIdToUri).toArray(String[]::new);
DatabaseConfig config = newDatabaseConfig(String.format("p%s", k), logFunction.apply(k), replicas);
return coordinator.<Database>getResource(config.getName(), config.resolve(clusterConfig)
.withSerializer(copycatConfig.getDefaultSerializer())
.withDefaultExecutor(copycatConfig.getDefaultExecutor()));
});
inMemoryDatabase = databases.remove(PartitionId.from(0));
partitionedDatabase = new PartitionedDatabase("onos-store", databases.values());
CompletableFuture<Void> status = coordinator.open()
.thenCompose(v -> CompletableFuture.allOf(inMemoryDatabase.open(), partitionedDatabase.open())
.whenComplete((db, error) -> {
if (error != null) {
log.error("Failed to initialize database.", error);
} else {
log.info("Successfully initialized database.");
}
}));
Futures.getUnchecked(status);
AsyncConsistentMap<TransactionId, Transaction> transactions =
this.<TransactionId, Transaction>consistentMapBuilder()
.withName("onos-transactions")
.withSerializer(Serializer.using(KryoNamespaces.API,
MapUpdate.class,
MapUpdate.Type.class,
Transaction.class,
Transaction.State.class))
.buildAsyncMap();
transactionManager = new TransactionManager(partitionedDatabase, transactions);
partitionedDatabase.setTransactionManager(transactionManager);
log.info("Started");
}
@Deactivate
public void deactivate() {
CompletableFuture.allOf(inMemoryDatabase.close(), partitionedDatabase.close())
.thenCompose(v -> coordinator.close())
.whenComplete((result, error) -> {
if (error != null) {
log.warn("Failed to cleanly close databases.", error);
} else {
log.info("Successfully closed databases.");
}
});
ImmutableList.copyOf(maps.values()).forEach(this::unregisterMap);
if (applicationService != null) {
applicationService.removeListener(appListener);
}
log.info("Stopped");
}
@Override
public TransactionContextBuilder transactionContextBuilder() {
return new DefaultTransactionContextBuilder(this::consistentMapBuilder,
transactionManager::execute,
transactionIdGenerator.get());
}
@Override
public List<PartitionInfo> getPartitionInfo() {
return Lists.asList(
inMemoryDatabase,
partitionedDatabase.getPartitions().toArray(new Database[]{}))
.stream()
.map(DatabaseManager::toPartitionInfo)
.collect(Collectors.toList());
}
private Log newPersistentLog() {
String logDir = System.getProperty("karaf.data", "./data");
return new FileLog()
.withDirectory(logDir)
.withSegmentSize(1073741824) // 1GB
.withFlushOnWrite(true)
.withSegmentInterval(Long.MAX_VALUE);
}
private Log newInMemoryLog() {
return new BufferedLog()
.withFlushOnWrite(false)
.withFlushInterval(Long.MAX_VALUE)
.withSegmentSize(10485760) // 10MB
.withSegmentInterval(Long.MAX_VALUE);
}
private DatabaseConfig newDatabaseConfig(String name, Log log, String[] replicas) {
return new DatabaseConfig()
.withName(name)
.withElectionTimeout(electionTimeoutMillis(replicas))
.withHeartbeatInterval(heartbeatTimeoutMillis(replicas))
.withConsistency(Consistency.DEFAULT)
.withLog(log)
.withDefaultSerializer(new DatabaseSerializer())
.withReplicas(replicas);
}
private long electionTimeoutMillis(String[] replicas) {
return replicas.length == 1 ? 10L : RAFT_ELECTION_TIMEOUT_MILLIS;
}
private long heartbeatTimeoutMillis(String[] replicas) {
return electionTimeoutMillis(replicas) / 2;
}
/**
* Maps a Raft Database object to a PartitionInfo object.
*
* @param database database containing input data
* @return PartitionInfo object
*/
private static PartitionInfo toPartitionInfo(Database database) {
return new PartitionInfo(database.name(),
database.cluster().term(),
database.cluster().members()
.stream()
.filter(member -> Type.ACTIVE.equals(member.type()))
.map(Member::uri)
.sorted()
.collect(Collectors.toList()),
database.cluster().leader() != null ?
database.cluster().leader().uri() : null);
}
@Override
public <K, V> EventuallyConsistentMapBuilder<K, V> eventuallyConsistentMapBuilder() {
return new EventuallyConsistentMapBuilderImpl<>(clusterService,
clusterCommunicator,
persistenceService);
}
@Override
public <K, V> ConsistentMapBuilder<K, V> consistentMapBuilder() {
return new DefaultConsistentMapBuilder<>(this);
}
@Override
public <E> DistributedSetBuilder<E> setBuilder() {
return new DefaultDistributedSetBuilder<>(() -> this.<E, Boolean>consistentMapBuilder());
}
@Override
public <E> DistributedQueueBuilder<E> queueBuilder() {
return new DefaultDistributedQueueBuilder<>(this);
}
@Override
public AtomicCounterBuilder atomicCounterBuilder() {
return new DefaultAtomicCounterBuilder(inMemoryDatabase, partitionedDatabase);
}
@Override
public <V> AtomicValueBuilder<V> atomicValueBuilder() {
Supplier<ConsistentMapBuilder<String, byte[]>> mapBuilderSupplier =
() -> this.<String, byte[]>consistentMapBuilder()
.withName("onos-atomic-values")
.withMeteringDisabled()
.withSerializer(Serializer.using(KryoNamespaces.BASIC));
return new DefaultAtomicValueBuilder<>(mapBuilderSupplier);
}
@Override
public LeaderElectorBuilder leaderElectorBuilder() {
throw new UnsupportedOperationException();
}
@Override
public List<MapInfo> getMapInfo() {
List<MapInfo> maps = Lists.newArrayList();
maps.addAll(getMapInfo(inMemoryDatabase));
maps.addAll(getMapInfo(partitionedDatabase));
return maps;
}
private List<MapInfo> getMapInfo(Database database) {
return complete(database.maps())
.stream()
.map(name -> new MapInfo(name, complete(database.mapSize(name))))
.filter(info -> info.size() > 0)
.collect(Collectors.toList());
}
@Override
public Map<String, Long> getCounters() {
Map<String, Long> counters = Maps.newHashMap();
counters.putAll(complete(inMemoryDatabase.counters()));
counters.putAll(complete(partitionedDatabase.counters()));
return counters;
}
@Override
public Map<String, Long> getPartitionedDatabaseCounters() {
Map<String, Long> counters = Maps.newHashMap();
counters.putAll(complete(partitionedDatabase.counters()));
return counters;
}
@Override
public Map<String, Long> getInMemoryDatabaseCounters() {
Map<String, Long> counters = Maps.newHashMap();
counters.putAll(complete(inMemoryDatabase.counters()));
return counters;
}
@Override
public Collection<TransactionId> getPendingTransactions() {
return complete(transactionManager.getPendingTransactionIds());
}
private static <T> T complete(CompletableFuture<T> future) {
try {
return future.get(DATABASE_OPERATION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ConsistentMapException.Interrupted();
} catch (TimeoutException e) {
throw new ConsistentMapException.Timeout();
} catch (ExecutionException e) {
throw new ConsistentMapException(e.getCause());
}
}
protected <K, V> DefaultAsyncConsistentMap<K, V> registerMap(DefaultAsyncConsistentMap<K, V> map) {
maps.put(map.name(), map);
if (map.applicationId() != null) {
mapsByApplication.put(map.applicationId(), map);
}
return map;
}
protected <K, V> void unregisterMap(DefaultAsyncConsistentMap<K, V> map) {
maps.remove(map.name(), map);
if (map.applicationId() != null) {
mapsByApplication.remove(map.applicationId(), map);
}
}
private class InternalApplicationListener implements ApplicationListener {
@Override
public void event(ApplicationEvent event) {
if (event.type() == APP_UNINSTALLED || event.type() == APP_DEACTIVATED) {
ApplicationId appId = event.subject().id();
List<DefaultAsyncConsistentMap> mapsToRemove;
synchronized (mapsByApplication) {
mapsToRemove = ImmutableList.copyOf(mapsByApplication.get(appId));
}
mapsToRemove.forEach(DatabaseManager.this::unregisterMap);
if (event.type() == APP_UNINSTALLED) {
mapsToRemove.stream().filter(map -> map.purgeOnUninstall()).forEach(map -> map.clear());
}
}
}
}
}

45
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabasePartitioner.java
View File

@ -1,45 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static com.google.common.base.Preconditions.checkState;
import java.util.List;
import com.google.common.base.Charsets;
import com.google.common.collect.ImmutableList;
import com.google.common.hash.Hashing;
/**
* Partitioner for mapping map entries to individual database partitions.
* <p>
* By default a md5 hash of the hash key (key or map name) is used to pick a
* partition.
*/
public abstract class DatabasePartitioner implements Partitioner<String> {
// Database partitions sorted by their partition name.
protected final List<Database> partitions;
public DatabasePartitioner(List<Database> partitions) {
checkState(partitions != null && !partitions.isEmpty(), "Partitions cannot be null or empty");
this.partitions = ImmutableList.copyOf(partitions);
}
protected int hash(String key) {
return Math.abs(Hashing.md5().newHasher().putBytes(key.getBytes(Charsets.UTF_8)).hash().asInt());
}
}

249
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseProxy.java
View File

@ -1,249 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import org.onlab.util.Match;
import org.onosproject.store.service.Versioned;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
/**
* Database proxy.
*/
public interface DatabaseProxy<K, V> {
/**
* Returns a set of all map names.
*
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Set<String>> maps();
/**
* Returns a mapping from counter name to next value.
*
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Map<String, Long>> counters();
/**
* Returns the number of entries in map.
*
* @param mapName map name
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Integer> mapSize(String mapName);
/**
* Checks whether the map is empty.
*
* @param mapName map name
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Boolean> mapIsEmpty(String mapName);
/**
* Checks whether the map contains a key.
*
* @param mapName map name
* @param key key to check.
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Boolean> mapContainsKey(String mapName, K key);
/**
* Checks whether the map contains a value.
*
* @param mapName map name
* @param value The value to check.
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Boolean> mapContainsValue(String mapName, V value);
/**
* Gets a value from the map.
*
* @param mapName map name
* @param key The key to get.
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Versioned<V>> mapGet(String mapName, K key);
/**
* Updates the map.
*
* @param mapName map name
* @param key The key to set
* @param valueMatch match for checking existing value
* @param versionMatch match for checking existing version
* @param value new value
* @return A completable future to be completed with the result once complete
*/
CompletableFuture<Result<UpdateResult<K, V>>> mapUpdate(
String mapName, K key, Match<V> valueMatch, Match<Long> versionMatch, V value);
/**
* Clears the map.
*
* @param mapName map name
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Result<Void>> mapClear(String mapName);
/**
* Gets a set of keys in the map.
*
* @param mapName map name
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Set<K>> mapKeySet(String mapName);
/**
* Gets a collection of values in the map.
*
* @param mapName map name
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Collection<Versioned<V>>> mapValues(String mapName);
/**
* Gets a set of entries in the map.
*
* @param mapName map name
* @return A completable future to be completed with the result once complete.
*/
CompletableFuture<Set<Map.Entry<K, Versioned<V>>>> mapEntrySet(String mapName);
/**
* Atomically add the given value to current value of the specified counter.
*
* @param counterName counter name
* @param delta value to add
* @return updated value
*/
CompletableFuture<Long> counterAddAndGet(String counterName, long delta);
/**
* Atomically add the given value to current value of the specified counter.
*
* @param counterName counter name
* @param delta value to add
* @return previous value
*/
CompletableFuture<Long> counterGetAndAdd(String counterName, long delta);
/**
* Atomically sets the given value to current value of the specified counter.
*
* @param counterName counter name
* @param value value to set
* @return void future
*/
CompletableFuture<Void> counterSet(String counterName, long value);
/**
* Atomically sets the given counter to the specified update value if and only if the current value is equal to the
* expected value.
* @param counterName counter name
* @param expectedValue value to use for equivalence check
* @param update value to set if expected value is current value
* @return true if an update occurred, false otherwise
*/
CompletableFuture<Boolean> counterCompareAndSet(String counterName, long expectedValue, long update);
/**
* Returns the current value of the specified atomic counter.
*
* @param counterName counter name
* @return current value
*/
CompletableFuture<Long> counterGet(String counterName);
/**
* Returns the size of queue.
*
* @param queueName queue name
* @return queue size
*/
CompletableFuture<Long> queueSize(String queueName);
/**
* Inserts an entry into the queue.
*
* @param queueName queue name
* @param entry queue entry
* @return void future
*/
CompletableFuture<Void> queuePush(String queueName, byte[] entry);
/**
* Removes an entry from the queue if the queue is non-empty.
*
* @param queueName queue name
* @return entry future. Can be completed with null if queue is empty
*/
CompletableFuture<byte[]> queuePop(String queueName);
/**
* Returns but does not remove an entry from the queue.
*
* @param queueName queue name
* @return entry. Can be null if queue is empty
*/
CompletableFuture<byte[]> queuePeek(String queueName);
/**
* Prepare and commit the specified transaction.
*
* @param transaction transaction to commit (after preparation)
* @return A completable future to be completed with the result once complete
*/
CompletableFuture<CommitResponse> prepareAndCommit(Transaction transaction);
/**
* Prepare the specified transaction for commit. A successful prepare implies
* all the affected resources are locked thus ensuring no concurrent updates can interfere.
*
* @param transaction transaction to prepare (for commit)
* @return A completable future to be completed with the result once complete. The future is completed
* with true if the transaction is successfully prepared i.e. all pre-conditions are met and
* applicable resources locked.
*/
CompletableFuture<Boolean> prepare(Transaction transaction);
/**
* Commit the specified transaction. A successful commit implies
* all the updates are applied, are now durable and are now visible externally.
*
* @param transaction transaction to commit
* @return A completable future to be completed with the result once complete
*/
CompletableFuture<CommitResponse> commit(Transaction transaction);
/**
* Rollback the specified transaction. A successful rollback implies
* all previously acquired locks for the affected resources are released.
*
* @param transaction transaction to rollback
* @return A completable future to be completed with the result once complete
*/
CompletableFuture<Boolean> rollback(Transaction transaction);
}

105
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseSerializer.java
View File

@ -1,105 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.nio.ByteBuffer;
import org.onlab.util.KryoNamespace;
import org.onlab.util.Match;
import org.onosproject.cluster.NodeId;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.serializers.KryoNamespaces;
import org.onosproject.store.serializers.KryoSerializer;
import org.onosproject.store.service.Versioned;
import net.kuujo.copycat.cluster.internal.MemberInfo;
import net.kuujo.copycat.raft.protocol.AppendRequest;
import net.kuujo.copycat.raft.protocol.AppendResponse;
import net.kuujo.copycat.raft.protocol.CommitRequest;
import net.kuujo.copycat.raft.protocol.CommitResponse;
import net.kuujo.copycat.raft.protocol.PollRequest;
import net.kuujo.copycat.raft.protocol.PollResponse;
import net.kuujo.copycat.raft.protocol.QueryRequest;
import net.kuujo.copycat.raft.protocol.QueryResponse;
import net.kuujo.copycat.raft.protocol.ReplicaInfo;
import net.kuujo.copycat.raft.protocol.SyncRequest;
import net.kuujo.copycat.raft.protocol.SyncResponse;
import net.kuujo.copycat.raft.protocol.VoteRequest;
import net.kuujo.copycat.raft.protocol.VoteResponse;
import net.kuujo.copycat.util.serializer.SerializerConfig;
/**
* Serializer for DatabaseManager's interaction with Copycat.
*/
public class DatabaseSerializer extends SerializerConfig {
private static final KryoNamespace COPYCAT = KryoNamespace.newBuilder()
.nextId(KryoNamespace.FLOATING_ID)
.register(AppendRequest.class)
.register(AppendResponse.class)
.register(SyncRequest.class)
.register(SyncResponse.class)
.register(VoteRequest.class)
.register(VoteResponse.class)
.register(PollRequest.class)
.register(PollResponse.class)
.register(QueryRequest.class)
.register(QueryResponse.class)
.register(CommitRequest.class)
.register(CommitResponse.class)
.register(ReplicaInfo.class)
.register(MemberInfo.class)
.build();
private static final KryoNamespace ONOS_STORE = KryoNamespace.newBuilder()
.nextId(KryoNamespace.FLOATING_ID)
.register(Versioned.class)
.register(MapUpdate.class)
.register(MapUpdate.Type.class)
.register(Result.class)
.register(UpdateResult.class)
.register(Result.Status.class)
.register(Transaction.class)
.register(Transaction.State.class)
.register(TransactionId.class)
.register(org.onosproject.store.primitives.impl.CommitResponse.class)
.register(Match.class)
.register(NodeId.class)
.build();
private static final KryoSerializer SERIALIZER = new KryoSerializer() {
@Override
protected void setupKryoPool() {
serializerPool = KryoNamespace.newBuilder()
.register(KryoNamespaces.BASIC)
.register(COPYCAT)
.register(ONOS_STORE)
.build();
}
};
@Override
public ByteBuffer writeObject(Object object) {
return ByteBuffer.wrap(SERIALIZER.encode(object));
}
@Override
public <T> T readObject(ByteBuffer buffer) {
return SERIALIZER.decode(buffer);
}
}

120
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java
View File

@ -1,120 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import net.kuujo.copycat.state.Command;
import net.kuujo.copycat.state.Initializer;
import net.kuujo.copycat.state.Query;
import net.kuujo.copycat.state.StateContext;
import org.onlab.util.Match;
import org.onosproject.store.service.Versioned;
import java.util.Collection;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
/**
* Database state.
*
*/
public interface DatabaseState<K, V> {
/**
* Initializes the database state.
*
* @param context The map state context.
*/
@Initializer
void init(StateContext<DatabaseState<K, V>> context);
@Query
Set<String> maps();
@Query
Map<String, Long> counters();
@Query
int mapSize(String mapName);
@Query
boolean mapIsEmpty(String mapName);
@Query
boolean mapContainsKey(String mapName, K key);
@Query
boolean mapContainsValue(String mapName, V value);
@Query
Versioned<V> mapGet(String mapName, K key);
@Command
Result<UpdateResult<K, V>> mapUpdate(String mapName, K key, Match<V> valueMatch, Match<Long> versionMatch, V value);
@Command
Result<Void> mapClear(String mapName);
@Query
Set<K> mapKeySet(String mapName);
@Query
Collection<Versioned<V>> mapValues(String mapName);
@Query
Set<Entry<K, Versioned<V>>> mapEntrySet(String mapName);
@Command
Long counterAddAndGet(String counterName, long delta);
@Command
Boolean counterCompareAndSet(String counterName, long expectedValue, long updateValue);
@Command
Long counterGetAndAdd(String counterName, long delta);
@Query
Long queueSize(String queueName);
@Query
byte[] queuePeek(String queueName);
@Command
byte[] queuePop(String queueName);
@Command
void queuePush(String queueName, byte[] entry);
@Query
Long counterGet(String counterName);
@Command
void counterSet(String counterName, long value);
@Command
CommitResponse prepareAndCommit(Transaction transaction);
@Command
boolean prepare(Transaction transaction);
@Command
CommitResponse commit(Transaction transaction);
@Command
boolean rollback(Transaction transaction);
}

107
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicCounter.java
View File

@ -1,107 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import org.onosproject.store.service.AsyncAtomicCounter;
import org.onosproject.utils.MeteringAgent;
import java.util.concurrent.CompletableFuture;
import static com.google.common.base.Preconditions.checkNotNull;
/**
* Default implementation for a distributed AsyncAtomicCounter backed by
* partitioned Raft DB.
* <p>
* The initial value will be zero.
*/
public class DefaultAsyncAtomicCounter implements AsyncAtomicCounter {
private final String name;
private final Database database;
private final MeteringAgent monitor;
private static final String PRIMITIVE_NAME = "atomicCounter";
private static final String INCREMENT_AND_GET = "incrementAndGet";
private static final String GET_AND_INCREMENT = "getAndIncrement";
private static final String GET_AND_ADD = "getAndAdd";
private static final String ADD_AND_GET = "addAndGet";
private static final String GET = "get";
private static final String SET = "set";
private static final String COMPARE_AND_SET = "compareAndSet";
public DefaultAsyncAtomicCounter(String name,
Database database,
boolean meteringEnabled) {
this.name = checkNotNull(name);
this.database = checkNotNull(database);
this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
}
@Override
public String name() {
return name;
}
@Override
public CompletableFuture<Long> incrementAndGet() {
final MeteringAgent.Context timer = monitor.startTimer(INCREMENT_AND_GET);
return addAndGet(1L)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Long> get() {
final MeteringAgent.Context timer = monitor.startTimer(GET);
return database.counterGet(name)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Long> getAndIncrement() {
final MeteringAgent.Context timer = monitor.startTimer(GET_AND_INCREMENT);
return getAndAdd(1L)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Long> getAndAdd(long delta) {
final MeteringAgent.Context timer = monitor.startTimer(GET_AND_ADD);
return database.counterGetAndAdd(name, delta)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Long> addAndGet(long delta) {
final MeteringAgent.Context timer = monitor.startTimer(ADD_AND_GET);
return database.counterAddAndGet(name, delta)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Void> set(long value) {
final MeteringAgent.Context timer = monitor.startTimer(SET);
return database.counterSet(name, value)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Boolean> compareAndSet(long expectedValue, long updateValue) {
final MeteringAgent.Context timer = monitor.startTimer(COMPARE_AND_SET);
return database.counterCompareAndSet(name, expectedValue, updateValue)
.whenComplete((r, e) -> timer.stop(e));
}
}

523
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncConsistentMap.java
View File

@ -1,523 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.onosproject.store.primitives.impl.StateMachineUpdate.Target.MAP_UPDATE;
import static org.onosproject.store.primitives.impl.StateMachineUpdate.Target.TX_COMMIT;
import static org.slf4j.LoggerFactory.getLogger;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.onlab.util.HexString;
import org.onlab.util.Match;
import org.onlab.util.SharedExecutors;
import org.onlab.util.Tools;
import org.onosproject.core.ApplicationId;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.service.AsyncConsistentMap;
import org.onosproject.store.service.ConsistentMapException;
import org.onosproject.store.service.ConsistentMapException.ConcurrentModification;
import org.onosproject.store.service.MapEvent;
import org.onosproject.store.service.MapEventListener;
import org.onosproject.store.service.MapTransaction;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.Versioned;
import org.onosproject.utils.MeteringAgent;
import org.slf4j.Logger;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Maps;
/**
* AsyncConsistentMap implementation that is backed by a Raft consensus
* based database.
*
* @param <K> type of key.
* @param <V> type of value.
*/
public class DefaultAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, V> {
private final String name;
private final ApplicationId applicationId;
private final Database database;
private final Serializer serializer;
private final boolean readOnly;
private final boolean purgeOnUninstall;
private static final String PRIMITIVE_NAME = "consistentMap";
private static final String SIZE = "size";
private static final String IS_EMPTY = "isEmpty";
private static final String CONTAINS_KEY = "containsKey";
private static final String CONTAINS_VALUE = "containsValue";
private static final String GET = "get";
private static final String COMPUTE_IF = "computeIf";
private static final String PUT = "put";
private static final String PUT_AND_GET = "putAndGet";
private static final String PUT_IF_ABSENT = "putIfAbsent";
private static final String REMOVE = "remove";
private static final String CLEAR = "clear";
private static final String KEY_SET = "keySet";
private static final String VALUES = "values";
private static final String ENTRY_SET = "entrySet";
private static final String REPLACE = "replace";
private static final String COMPUTE_IF_ABSENT = "computeIfAbsent";
private final Set<MapEventListener<K, V>> listeners = new CopyOnWriteArraySet<>();
private final Logger log = getLogger(getClass());
private final MeteringAgent monitor;
private static final String ERROR_NULL_KEY = "Key cannot be null";
private static final String ERROR_NULL_VALUE = "Null values are not allowed";
// String representation of serialized byte[] -> original key Object
private final LoadingCache<String, K> keyCache = CacheBuilder.newBuilder()
.softValues()
.build(new CacheLoader<String, K>() {
@Override
public K load(String key) {
return serializer.decode(HexString.fromHexString(key));
}
});
protected String sK(K key) {
String s = HexString.toHexString(serializer.encode(key));
keyCache.put(s, key);
return s;
}
protected K dK(String key) {
return keyCache.getUnchecked(key);
}
public DefaultAsyncConsistentMap(String name,
ApplicationId applicationId,
Database database,
Serializer serializer,
boolean readOnly,
boolean purgeOnUninstall,
boolean meteringEnabled) {
this.name = checkNotNull(name, "map name cannot be null");
this.applicationId = applicationId;
this.database = checkNotNull(database, "database cannot be null");
this.serializer = checkNotNull(serializer, "serializer cannot be null");
this.readOnly = readOnly;
this.purgeOnUninstall = purgeOnUninstall;
this.database.registerConsumer(update -> {
SharedExecutors.getSingleThreadExecutor().execute(() -> {
if (listeners.isEmpty()) {
return;
}
try {
if (update.target() == MAP_UPDATE) {
Result<UpdateResult<String, byte[]>> result = update.output();
if (result.success() && result.value().mapName().equals(name)) {
MapEvent<K, V> mapEvent = result.value()
.<K, V>map(this::dK,
v -> serializer.decode(Tools.copyOf(v)))
.toMapEvent();
notifyListeners(mapEvent);
}
} else if (update.target() == TX_COMMIT) {
CommitResponse response = update.output();
if (response.success()) {
response.updates().forEach(u -> {
if (u.mapName().equals(name)) {
MapEvent<K, V> mapEvent =
u.<K, V>map(this::dK,
v -> serializer.decode(Tools.copyOf(v)))
.toMapEvent();
notifyListeners(mapEvent);
}
});
}
}
} catch (Exception e) {
log.warn("Error notifying listeners", e);
}
});
});
this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
}
/**
* Returns this map name.
* @return map name
*/
@Override
public String name() {
return name;
}
/**
* Returns the serializer for map entries.
* @return map entry serializer
*/
public Serializer serializer() {
return serializer;
}
/**
* Returns the applicationId owning this map.
* @return application Id
*/
@Override
public ApplicationId applicationId() {
return applicationId;
}
/**
* Returns whether the map entries should be purged when the application
* owning it is uninstalled.
* @return true is map needs to cleared on app uninstall; false otherwise
*/
public boolean purgeOnUninstall() {
return purgeOnUninstall;
}
@Override
public CompletableFuture<Integer> size() {
final MeteringAgent.Context timer = monitor.startTimer(SIZE);
return database.mapSize(name)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Boolean> isEmpty() {
final MeteringAgent.Context timer = monitor.startTimer(IS_EMPTY);
return database.mapIsEmpty(name)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Boolean> containsKey(K key) {
checkNotNull(key, ERROR_NULL_KEY);
final MeteringAgent.Context timer = monitor.startTimer(CONTAINS_KEY);
return database.mapContainsKey(name, sK(key))
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Boolean> containsValue(V value) {
checkNotNull(value, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(CONTAINS_VALUE);
return database.mapContainsValue(name, serializer.encode(value))
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Versioned<V>> get(K key) {
checkNotNull(key, ERROR_NULL_KEY);
final MeteringAgent.Context timer = monitor.startTimer(GET);
return database.mapGet(name, sK(key))
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v != null ? v.map(serializer::decode) : null);
}
@Override
public CompletableFuture<Versioned<V>> computeIfAbsent(K key,
Function<? super K, ? extends V> mappingFunction) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(mappingFunction, "Mapping function cannot be null");
final MeteringAgent.Context timer = monitor.startTimer(COMPUTE_IF_ABSENT);
return updateAndGet(key, Match.ifNull(), Match.any(), mappingFunction.apply(key))
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.newValue());
}
@Override
public CompletableFuture<Versioned<V>> computeIfPresent(K key,
BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
return computeIf(key, Objects::nonNull, remappingFunction);
}
@Override
public CompletableFuture<Versioned<V>> compute(K key,
BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
return computeIf(key, v -> true, remappingFunction);
}
@Override
public CompletableFuture<Versioned<V>> computeIf(K key,
Predicate<? super V> condition,
BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(condition, "predicate function cannot be null");
checkNotNull(remappingFunction, "Remapping function cannot be null");
final MeteringAgent.Context timer = monitor.startTimer(COMPUTE_IF);
return get(key).thenCompose(r1 -> {
V existingValue = r1 == null ? null : r1.value();
// if the condition evaluates to false, return existing value.
if (!condition.test(existingValue)) {
return CompletableFuture.completedFuture(r1);
}
AtomicReference<V> computedValue = new AtomicReference<>();
// if remappingFunction throws an exception, return the exception.
try {
computedValue.set(remappingFunction.apply(key, existingValue));
} catch (Exception e) {
return Tools.exceptionalFuture(e);
}
if (computedValue.get() == null && r1 == null) {
return CompletableFuture.completedFuture(null);
}
Match<V> valueMatcher = r1 == null ? Match.ifNull() : Match.any();
Match<Long> versionMatcher = r1 == null ? Match.any() : Match.ifValue(r1.version());
return updateAndGet(key, valueMatcher, versionMatcher, computedValue.get())
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> {
if (v.updated()) {
return v.newValue();
} else {
throw new ConcurrentModification("Concurrent update to " + name + " detected");
}
});
});
}
@Override
public CompletableFuture<Versioned<V>> put(K key, V value) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(value, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(PUT);
return updateAndGet(key, Match.any(), Match.any(), value).thenApply(v -> v.oldValue())
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Versioned<V>> putAndGet(K key, V value) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(value, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(PUT_AND_GET);
return updateAndGet(key, Match.any(), Match.any(), value).thenApply(v -> v.newValue())
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Versioned<V>> remove(K key) {
checkNotNull(key, ERROR_NULL_KEY);
final MeteringAgent.Context timer = monitor.startTimer(REMOVE);
return updateAndGet(key, Match.any(), Match.any(), null).thenApply(v -> v.oldValue())
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Void> clear() {
checkIfUnmodifiable();
final MeteringAgent.Context timer = monitor.startTimer(CLEAR);
return database.mapClear(name).thenApply(this::unwrapResult)
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Set<K>> keySet() {
final MeteringAgent.Context timer = monitor.startTimer(KEY_SET);
return database.mapKeySet(name)
.thenApply(s -> newMappingKeySet(s))
.whenComplete((r, e) -> timer.stop(e));
}
@Override
public CompletableFuture<Collection<Versioned<V>>> values() {
final MeteringAgent.Context timer = monitor.startTimer(VALUES);
return database.mapValues(name)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(c -> c
.stream()
.map(v -> v.<V>map(serializer::decode))
.collect(Collectors.toList()));
}
@Override
public CompletableFuture<Set<Entry<K, Versioned<V>>>> entrySet() {
final MeteringAgent.Context timer = monitor.startTimer(ENTRY_SET);
return database.mapEntrySet(name)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(s -> newMappingEntrySet(s));
}
@Override
public CompletableFuture<Versioned<V>> putIfAbsent(K key, V value) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(value, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(PUT_IF_ABSENT);
return updateAndGet(key, Match.ifNull(), Match.any(), value)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.oldValue());
}
@Override
public CompletableFuture<Boolean> remove(K key, V value) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(value, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(REMOVE);
return updateAndGet(key, Match.ifValue(value), Match.any(), null)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.updated());
}
@Override
public CompletableFuture<Boolean> remove(K key, long version) {
checkNotNull(key, ERROR_NULL_KEY);
final MeteringAgent.Context timer = monitor.startTimer(REMOVE);
return updateAndGet(key, Match.any(), Match.ifValue(version), null)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.updated());
}
@Override
public CompletableFuture<Versioned<V>> replace(K key, V value) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(value, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(REPLACE);
return updateAndGet(key, Match.ifNotNull(), Match.any(), value)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.oldValue());
}
@Override
public CompletableFuture<Boolean> replace(K key, V oldValue, V newValue) {
checkNotNull(key, ERROR_NULL_KEY);
checkNotNull(oldValue, ERROR_NULL_VALUE);
checkNotNull(newValue, ERROR_NULL_VALUE);
final MeteringAgent.Context timer = monitor.startTimer(REPLACE);
return updateAndGet(key, Match.ifValue(oldValue), Match.any(), newValue)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.updated());
}
@Override
public CompletableFuture<Boolean> replace(K key, long oldVersion, V newValue) {
final MeteringAgent.Context timer = monitor.startTimer(REPLACE);
return updateAndGet(key, Match.any(), Match.ifValue(oldVersion), newValue)
.whenComplete((r, e) -> timer.stop(e))
.thenApply(v -> v.updated());
}
/**
* Pre-update hook for performing required checks/actions before going forward with an update operation.
* @param key map key.
*/
protected void beforeUpdate(K key) {
checkIfUnmodifiable();
}
private Set<K> newMappingKeySet(Set<String> s) {
return new MappingSet<>(s, Collections::unmodifiableSet,
this::sK, this::dK);
}
private Set<Entry<K, Versioned<V>>> newMappingEntrySet(Set<Entry<String, Versioned<byte[]>>> s) {
return new MappingSet<>(s, Collections::unmodifiableSet,
this::reverseMapRawEntry, this::mapRawEntry);
}
private Map.Entry<K, Versioned<V>> mapRawEntry(Map.Entry<String, Versioned<byte[]>> e) {
return Maps.immutableEntry(dK(e.getKey()), e.getValue().<V>map(serializer::decode));
}
private Map.Entry<String, Versioned<byte[]>> reverseMapRawEntry(Map.Entry<K, Versioned<V>> e) {
return Maps.immutableEntry(sK(e.getKey()), e.getValue().map(serializer::encode));
}
private CompletableFuture<UpdateResult<K, V>> updateAndGet(K key,
Match<V> oldValueMatch,
Match<Long> oldVersionMatch,
V value) {
beforeUpdate(key);
return database.mapUpdate(name,
sK(key),
oldValueMatch.map(serializer::encode),
oldVersionMatch,
value == null ? null : serializer.encode(value))
.thenApply(this::unwrapResult)
.thenApply(r -> r.<K, V>map(this::dK, serializer::decode));
}
private <T> T unwrapResult(Result<T> result) {
if (result.status() == Result.Status.LOCKED) {
throw new ConsistentMapException.ConcurrentModification();
} else if (result.success()) {
return result.value();
} else {
throw new IllegalStateException("Must not be here");
}
}
private void checkIfUnmodifiable() {
if (readOnly) {
throw new UnsupportedOperationException();
}
}
@Override
public CompletableFuture<Void> addListener(MapEventListener<K, V> listener) {
listeners.add(listener);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletableFuture<Void> removeListener(MapEventListener<K, V> listener) {
listeners.remove(listener);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
return Tools.exceptionalFuture(new UnsupportedOperationException());
}
@Override
public CompletableFuture<Void> commit(TransactionId transactionId) {
return Tools.exceptionalFuture(new UnsupportedOperationException());
}
@Override
public CompletableFuture<Void> rollback(TransactionId transactionId) {
return Tools.exceptionalFuture(new UnsupportedOperationException());
}
protected void notifyListeners(MapEvent<K, V> event) {
if (event == null) {
return;
}
listeners.forEach(listener -> {
try {
listener.event(event);
} catch (Exception e) {
log.warn("Failure notifying listener about {}", event, e);
}
});
}
}

41
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounterBuilder.java
View File

@ -1,41 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import org.onosproject.store.service.AsyncAtomicCounter;
import org.onosproject.store.service.AtomicCounterBuilder;
import static com.google.common.base.Preconditions.checkNotNull;
/**
* Default implementation of AtomicCounterBuilder.
*/
public class DefaultAtomicCounterBuilder extends AtomicCounterBuilder {
private final Database partitionedDatabase;
private final Database inMemoryDatabase;
public DefaultAtomicCounterBuilder(Database inMemoryDatabase, Database partitionedDatabase) {
this.inMemoryDatabase = inMemoryDatabase;
this.partitionedDatabase = partitionedDatabase;
}
@Override
public AsyncAtomicCounter build() {
Database database = partitionsDisabled() ? inMemoryDatabase : partitionedDatabase;
return new DefaultAsyncAtomicCounter(checkNotNull(name()), database, meteringEnabled());
}
}

79
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMapBuilder.java
View File

@ -1,79 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import org.onosproject.store.service.AsyncConsistentMap;
import org.onosproject.store.service.ConsistentMap;
import org.onosproject.store.service.ConsistentMapBuilder;
import static com.google.common.base.Preconditions.checkState;
/**
* Default Consistent Map builder.
*
* @param <K> type for map key
* @param <V> type for map value
*/
public class DefaultConsistentMapBuilder<K, V> extends ConsistentMapBuilder<K, V> {
private final DatabaseManager manager;
public DefaultConsistentMapBuilder(DatabaseManager manager) {
this.manager = manager;
}
private void validateInputs() {
checkState(name() != null, "name must be specified");
checkState(serializer() != null, "serializer must be specified");
if (purgeOnUninstall()) {
checkState(applicationId() != null, "ApplicationId must be specified when purgeOnUninstall is enabled");
}
}
@Override
public ConsistentMap<K, V> build() {
return buildAndRegisterMap().asConsistentMap();
}
@Override
public AsyncConsistentMap<K, V> buildAsyncMap() {
return buildAndRegisterMap();
}
private DefaultAsyncConsistentMap<K, V> buildAndRegisterMap() {
validateInputs();
Database database = partitionsDisabled() ? manager.inMemoryDatabase : manager.partitionedDatabase;
if (relaxedReadConsistency()) {
return manager.registerMap(
new AsyncCachingConsistentMap<>(name(),
applicationId(),
database,
serializer(),
readOnly(),
purgeOnUninstall(),
meteringEnabled()));
} else {
return manager.registerMap(
new DefaultAsyncConsistentMap<>(name(),
applicationId(),
database,
serializer(),
readOnly(),
purgeOnUninstall(),
meteringEnabled()));
}
}
}

253
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabase.java
View File

@ -1,253 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import com.google.common.collect.Sets;
import net.kuujo.copycat.resource.internal.AbstractResource;
import net.kuujo.copycat.resource.internal.ResourceManager;
import net.kuujo.copycat.state.StateMachine;
import net.kuujo.copycat.state.internal.DefaultStateMachine;
import net.kuujo.copycat.util.concurrent.Futures;
import net.kuujo.copycat.util.function.TriConsumer;
import org.onlab.util.Match;
import org.onosproject.store.service.Versioned;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Supplier;
/**
* Default database.
*/
public class DefaultDatabase extends AbstractResource<Database> implements Database {
private final StateMachine<DatabaseState<String, byte[]>> stateMachine;
private DatabaseProxy<String, byte[]> proxy;
private final Set<Consumer<StateMachineUpdate>> consumers = Sets.newCopyOnWriteArraySet();
private final TriConsumer<String, Object, Object> watcher = new InternalStateMachineWatcher();
@SuppressWarnings({"unchecked", "rawtypes"})
public DefaultDatabase(ResourceManager context) {
super(context);
this.stateMachine = new DefaultStateMachine(context,
DatabaseState.class,
DefaultDatabaseState.class,
DefaultDatabase.class.getClassLoader());
this.stateMachine.addStartupTask(() -> {
stateMachine.registerWatcher(watcher);
return CompletableFuture.completedFuture(null);
});
this.stateMachine.addShutdownTask(() -> {
stateMachine.unregisterWatcher(watcher);
return CompletableFuture.completedFuture(null);
});
}
/**
* If the database is closed, returning a failed CompletableFuture. Otherwise, calls the given supplier to
* return the completed future result.
*
* @param supplier The supplier to call if the database is open.
* @param <T> The future result type.
* @return A completable future that if this database is closed is immediately failed.
*/
protected <T> CompletableFuture<T> checkOpen(Supplier<CompletableFuture<T>> supplier) {
if (proxy == null) {
return Futures.exceptionalFuture(new IllegalStateException("Database closed"));
}
return supplier.get();
}
@Override
public CompletableFuture<Set<String>> maps() {
return checkOpen(() -> proxy.maps());
}
@Override
public CompletableFuture<Map<String, Long>> counters() {
return checkOpen(() -> proxy.counters());
}
@Override
public CompletableFuture<Integer> mapSize(String mapName) {
return checkOpen(() -> proxy.mapSize(mapName));
}
@Override
public CompletableFuture<Boolean> mapIsEmpty(String mapName) {
return checkOpen(() -> proxy.mapIsEmpty(mapName));
}
@Override
public CompletableFuture<Boolean> mapContainsKey(String mapName, String key) {
return checkOpen(() -> proxy.mapContainsKey(mapName, key));
}
@Override
public CompletableFuture<Boolean> mapContainsValue(String mapName, byte[] value) {
return checkOpen(() -> proxy.mapContainsValue(mapName, value));
}
@Override
public CompletableFuture<Versioned<byte[]>> mapGet(String mapName, String key) {
return checkOpen(() -> proxy.mapGet(mapName, key));
}
@Override
public CompletableFuture<Result<UpdateResult<String, byte[]>>> mapUpdate(
String mapName, String key, Match<byte[]> valueMatch, Match<Long> versionMatch, byte[] value) {
return checkOpen(() -> proxy.mapUpdate(mapName, key, valueMatch, versionMatch, value));
}
@Override
public CompletableFuture<Result<Void>> mapClear(String mapName) {
return checkOpen(() -> proxy.mapClear(mapName));
}
@Override
public CompletableFuture<Set<String>> mapKeySet(String mapName) {
return checkOpen(() -> proxy.mapKeySet(mapName));
}
@Override
public CompletableFuture<Collection<Versioned<byte[]>>> mapValues(String mapName) {
return checkOpen(() -> proxy.mapValues(mapName));
}
@Override
public CompletableFuture<Set<Map.Entry<String, Versioned<byte[]>>>> mapEntrySet(String mapName) {
return checkOpen(() -> proxy.mapEntrySet(mapName));
}
@Override
public CompletableFuture<Long> counterGet(String counterName) {
return checkOpen(() -> proxy.counterGet(counterName));
}
@Override
public CompletableFuture<Long> counterAddAndGet(String counterName, long delta) {
return checkOpen(() -> proxy.counterAddAndGet(counterName, delta));
}
@Override
public CompletableFuture<Long> counterGetAndAdd(String counterName, long delta) {
return checkOpen(() -> proxy.counterGetAndAdd(counterName, delta));
}
@Override
public CompletableFuture<Void> counterSet(String counterName, long value) {
return checkOpen(() -> proxy.counterSet(counterName, value));
}
@Override
public CompletableFuture<Boolean> counterCompareAndSet(String counterName, long expectedValue, long update) {
return checkOpen(() -> proxy.counterCompareAndSet(counterName, expectedValue, update));
}
@Override
public CompletableFuture<Long> queueSize(String queueName) {
return checkOpen(() -> proxy.queueSize(queueName));
}
@Override
public CompletableFuture<Void> queuePush(String queueName, byte[] entry) {
return checkOpen(() -> proxy.queuePush(queueName, entry));
}
@Override
public CompletableFuture<byte[]> queuePop(String queueName) {
return checkOpen(() -> proxy.queuePop(queueName));
}
@Override
public CompletableFuture<byte[]> queuePeek(String queueName) {
return checkOpen(() -> proxy.queuePeek(queueName));
}
@Override
public CompletableFuture<CommitResponse> prepareAndCommit(Transaction transaction) {
return checkOpen(() -> proxy.prepareAndCommit(transaction));
}
@Override
public CompletableFuture<Boolean> prepare(Transaction transaction) {
return checkOpen(() -> proxy.prepare(transaction));
}
@Override
public CompletableFuture<CommitResponse> commit(Transaction transaction) {
return checkOpen(() -> proxy.commit(transaction));
}
@Override
public CompletableFuture<Boolean> rollback(Transaction transaction) {
return checkOpen(() -> proxy.rollback(transaction));
}
@Override
@SuppressWarnings("unchecked")
public synchronized CompletableFuture<Database> open() {
return runStartupTasks()
.thenCompose(v -> stateMachine.open())
.thenRun(() -> {
this.proxy = stateMachine.createProxy(DatabaseProxy.class, this.getClass().getClassLoader());
})
.thenApply(v -> null);
}
@Override
public synchronized CompletableFuture<Void> close() {
proxy = null;
return stateMachine.close()
.thenCompose(v -> runShutdownTasks());
}
@Override
public int hashCode() {
return name().hashCode();
}
@Override
public boolean equals(Object other) {
if (other instanceof Database) {
return name().equals(((Database) other).name());
}
return false;
}
@Override
public void registerConsumer(Consumer<StateMachineUpdate> consumer) {
consumers.add(consumer);
}
@Override
public void unregisterConsumer(Consumer<StateMachineUpdate> consumer) {
consumers.remove(consumer);
}
private class InternalStateMachineWatcher implements TriConsumer<String, Object, Object> {
@Override
public void accept(String name, Object input, Object output) {
StateMachineUpdate update = new StateMachineUpdate(name, input, output);
consumers.forEach(consumer -> consumer.accept(update));
}
}
}

381
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java
View File

@ -1,381 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import net.kuujo.copycat.state.Initializer;
import net.kuujo.copycat.state.StateContext;
import org.onlab.util.Match;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.service.Versioned;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedList;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
/**
* Default database state.
*/
public class DefaultDatabaseState implements DatabaseState<String, byte[]> {
private Long nextVersion;
private Map<String, AtomicLong> counters;
private Map<String, Map<String, Versioned<byte[]>>> maps;
private Map<String, Queue<byte[]>> queues;
/**
* This locks map has a structure similar to the "tables" map above and
* holds all the provisional updates made during a transaction's prepare phase.
* The entry value is represented as the tuple: (transactionId, newValue)
* If newValue == null that signifies this update is attempting to
* delete the existing value.
* This map also serves as a lock on the entries that are being updated.
* The presence of a entry in this map indicates that element is
* participating in a transaction and is currently locked for updates.
*/
private Map<String, Map<String, Update>> locks;
@Initializer
@Override
public void init(StateContext<DatabaseState<String, byte[]>> context) {
counters = context.get("counters");
if (counters == null) {
counters = Maps.newConcurrentMap();
context.put("counters", counters);
}
maps = context.get("maps");
if (maps == null) {
maps = Maps.newConcurrentMap();
context.put("maps", maps);
}
locks = context.get("locks");
if (locks == null) {
locks = Maps.newConcurrentMap();
context.put("locks", locks);
}
queues = context.get("queues");
if (queues == null) {
queues = Maps.newConcurrentMap();
context.put("queues", queues);
}
nextVersion = context.get("nextVersion");
if (nextVersion == null) {
nextVersion = 0L;
context.put("nextVersion", nextVersion);
}
}
@Override
public Set<String> maps() {
return ImmutableSet.copyOf(maps.keySet());
}
@Override
public Map<String, Long> counters() {
Map<String, Long> counterMap = Maps.newHashMap();
counters.forEach((k, v) -> counterMap.put(k, v.get()));
return counterMap;
}
@Override
public int mapSize(String mapName) {
return getMap(mapName).size();
}
@Override
public boolean mapIsEmpty(String mapName) {
return getMap(mapName).isEmpty();
}
@Override
public boolean mapContainsKey(String mapName, String key) {
return getMap(mapName).containsKey(key);
}
@Override
public boolean mapContainsValue(String mapName, byte[] value) {
return getMap(mapName).values().stream().anyMatch(v -> Arrays.equals(v.value(), value));
}
@Override
public Versioned<byte[]> mapGet(String mapName, String key) {
return getMap(mapName).get(key);
}
@Override
public Result<UpdateResult<String, byte[]>> mapUpdate(
String mapName,
String key,
Match<byte[]> valueMatch,
Match<Long> versionMatch,
byte[] value) {
if (isLockedForUpdates(mapName, key)) {
return Result.locked();
}
Versioned<byte[]> currentValue = getMap(mapName).get(key);
if (!valueMatch.matches(currentValue == null ? null : currentValue.value()) ||
!versionMatch.matches(currentValue == null ? null : currentValue.version())) {
return Result.ok(new UpdateResult<>(false, mapName, key, currentValue, currentValue));
} else {
if (value == null) {
if (currentValue == null) {
return Result.ok(new UpdateResult<>(false, mapName, key, null, null));
} else {
getMap(mapName).remove(key);
return Result.ok(new UpdateResult<>(true, mapName, key, currentValue, null));
}
}
Versioned<byte[]> newValue = new Versioned<>(value, ++nextVersion);
getMap(mapName).put(key, newValue);
return Result.ok(new UpdateResult<>(true, mapName, key, currentValue, newValue));
}
}
@Override
public Result<Void> mapClear(String mapName) {
if (areTransactionsInProgress(mapName)) {
return Result.locked();
}
getMap(mapName).clear();
return Result.ok(null);
}
@Override
public Set<String> mapKeySet(String mapName) {
return ImmutableSet.copyOf(getMap(mapName).keySet());
}
@Override
public Collection<Versioned<byte[]>> mapValues(String mapName) {
return ImmutableList.copyOf(getMap(mapName).values());
}
@Override
public Set<Entry<String, Versioned<byte[]>>> mapEntrySet(String mapName) {
return ImmutableSet.copyOf(getMap(mapName)
.entrySet()
.stream()
.map(entry -> Maps.immutableEntry(entry.getKey(), entry.getValue()))
.collect(Collectors.toSet()));
}
@Override
public Long counterAddAndGet(String counterName, long delta) {
return getCounter(counterName).addAndGet(delta);
}
@Override
public Long counterGetAndAdd(String counterName, long delta) {
return getCounter(counterName).getAndAdd(delta);
}
@Override
public Boolean counterCompareAndSet(String counterName, long expectedValue, long updateValue) {
return getCounter(counterName).compareAndSet(expectedValue, updateValue);
}
@Override
public Long counterGet(String counterName) {
return getCounter(counterName).get();
}
@Override
public void counterSet(String counterName, long value) {
getCounter(counterName).set(value);
}
@Override
public Long queueSize(String queueName) {
return Long.valueOf(getQueue(queueName).size());
}
@Override
public byte[] queuePeek(String queueName) {
return getQueue(queueName).peek();
}
@Override
public byte[] queuePop(String queueName) {
return getQueue(queueName).poll();
}
@Override
public void queuePush(String queueName, byte[] entry) {
getQueue(queueName).offer(entry);
}
@Override
public CommitResponse prepareAndCommit(Transaction transaction) {
if (prepare(transaction)) {
return commit(transaction);
}
return CommitResponse.failure();
}
@Override
public boolean prepare(Transaction transaction) {
if (transaction.updates().stream().anyMatch(update ->
isLockedByAnotherTransaction(update.mapName(),
update.key(),
transaction.id()))) {
return false;
}
if (transaction.updates().stream().allMatch(this::isUpdatePossible)) {
transaction.updates().forEach(update -> doProvisionalUpdate(update, transaction.id()));
return true;
}
return false;
}
@Override
public CommitResponse commit(Transaction transaction) {
return CommitResponse.success(Lists.transform(transaction.updates(),
update -> commitProvisionalUpdate(update, transaction.id())));
}
@Override
public boolean rollback(Transaction transaction) {
transaction.updates().forEach(update -> undoProvisionalUpdate(update, transaction.id()));
return true;
}
private Map<String, Versioned<byte[]>> getMap(String mapName) {
return maps.computeIfAbsent(mapName, name -> Maps.newConcurrentMap());
}
private Map<String, Update> getLockMap(String mapName) {
return locks.computeIfAbsent(mapName, name -> Maps.newConcurrentMap());
}
private AtomicLong getCounter(String counterName) {
return counters.computeIfAbsent(counterName, name -> new AtomicLong(0));
}
private Queue<byte[]> getQueue(String queueName) {
return queues.computeIfAbsent(queueName, name -> new LinkedList<>());
}
private boolean isUpdatePossible(MapUpdate<String, byte[]> update) {
Versioned<byte[]> existingEntry = mapGet(update.mapName(), update.key());
switch (update.type()) {
case PUT:
case REMOVE:
return true;
case PUT_IF_ABSENT:
return existingEntry == null;
case PUT_IF_VERSION_MATCH:
return existingEntry != null && existingEntry.version() == update.currentVersion();
case PUT_IF_VALUE_MATCH:
return existingEntry != null && Arrays.equals(existingEntry.value(), update.currentValue());
case REMOVE_IF_VERSION_MATCH:
return existingEntry == null || existingEntry.version() == update.currentVersion();
case REMOVE_IF_VALUE_MATCH:
return existingEntry == null || Arrays.equals(existingEntry.value(), update.currentValue());
default:
throw new IllegalStateException("Unsupported type: " + update.type());
}
}
private void doProvisionalUpdate(MapUpdate<String, byte[]> update, TransactionId transactionId) {
Map<String, Update> lockMap = getLockMap(update.mapName());
switch (update.type()) {
case PUT:
case PUT_IF_ABSENT:
case PUT_IF_VERSION_MATCH:
case PUT_IF_VALUE_MATCH:
lockMap.put(update.key(), new Update(transactionId, update.value()));
break;
case REMOVE:
case REMOVE_IF_VERSION_MATCH:
case REMOVE_IF_VALUE_MATCH:
lockMap.put(update.key(), new Update(transactionId, null));
break;
default:
throw new IllegalStateException("Unsupported type: " + update.type());
}
}
private UpdateResult<String, byte[]> commitProvisionalUpdate(
MapUpdate<String, byte[]> update, TransactionId transactionId) {
String mapName = update.mapName();
String key = update.key();
Update provisionalUpdate = getLockMap(mapName).get(key);
if (Objects.equal(transactionId, provisionalUpdate.transactionId())) {
getLockMap(mapName).remove(key);
} else {
throw new IllegalStateException("Invalid transaction Id");
}
return mapUpdate(mapName, key, Match.any(), Match.any(), provisionalUpdate.value()).value();
}
private void undoProvisionalUpdate(MapUpdate<String, byte[]> update, TransactionId transactionId) {
String mapName = update.mapName();
String key = update.key();
Update provisionalUpdate = getLockMap(mapName).get(key);
if (provisionalUpdate == null) {
return;
}
if (Objects.equal(transactionId, provisionalUpdate.transactionId())) {
getLockMap(mapName).remove(key);
}
}
private boolean isLockedByAnotherTransaction(String mapName, String key, TransactionId transactionId) {
Update update = getLockMap(mapName).get(key);
return update != null && !Objects.equal(transactionId, update.transactionId());
}
private boolean isLockedForUpdates(String mapName, String key) {
return getLockMap(mapName).containsKey(key);
}
private boolean areTransactionsInProgress(String mapName) {
return !getLockMap(mapName).isEmpty();
}
private class Update {
private final TransactionId transactionId;
private final byte[] value;
public Update(TransactionId txId, byte[] value) {
this.transactionId = txId;
this.value = value;
}
public TransactionId transactionId() {
return this.transactionId;
}
public byte[] value() {
return this.value;
}
}
}

138
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueue.java
View File

@ -1,138 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Futures;
import org.onlab.util.SharedExecutors;
import org.onosproject.store.service.DistributedPrimitive;
import org.onosproject.store.service.DistributedQueue;
import org.onosproject.store.service.Serializer;
import org.onosproject.utils.MeteringAgent;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.onosproject.store.primitives.impl.StateMachineUpdate.Target.QUEUE_PUSH;
/**
* DistributedQueue implementation that provides FIFO ordering semantics.
*
* @param <E> queue entry type
*/
public class DefaultDistributedQueue<E> implements DistributedQueue<E> {
private final String name;
private final Database database;
private final Serializer serializer;
private final Set<CompletableFuture<E>> pendingFutures = Sets.newIdentityHashSet();
private static final String PRIMITIVE_NAME = "distributedQueue";
private static final String SIZE = "size";
private static final String PUSH = "push";
private static final String POP = "pop";
private static final String PEEK = "peek";
private static final String ERROR_NULL_ENTRY = "Null entries are not allowed";
private final MeteringAgent monitor;
public DefaultDistributedQueue(String name,
Database database,
Serializer serializer,
boolean meteringEnabled) {
this.name = checkNotNull(name, "queue name cannot be null");
this.database = checkNotNull(database, "database cannot be null");
this.serializer = checkNotNull(serializer, "serializer cannot be null");
this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
this.database.registerConsumer(update -> {
SharedExecutors.getSingleThreadExecutor().execute(() -> {
if (update.target() == QUEUE_PUSH) {
List<Object> input = update.input();
String queueName = (String) input.get(0);
if (queueName.equals(name)) {
tryPoll();
}
}
});
});
}
@Override
public long size() {
final MeteringAgent.Context timer = monitor.startTimer(SIZE);
return Futures.getUnchecked(database.queueSize(name).whenComplete((r, e) -> timer.stop(e)));
}
@Override
public void push(E entry) {
checkNotNull(entry, ERROR_NULL_ENTRY);
final MeteringAgent.Context timer = monitor.startTimer(PUSH);
Futures.getUnchecked(database.queuePush(name, serializer.encode(entry))
.whenComplete((r, e) -> timer.stop(e)));
}
@Override
public CompletableFuture<E> pop() {
final MeteringAgent.Context timer = monitor.startTimer(POP);
return database.queuePop(name)
.whenComplete((r, e) -> timer.stop(e))
.thenCompose(v -> {
if (v != null) {
return CompletableFuture.<E>completedFuture(serializer.decode(v));
}
CompletableFuture<E> newPendingFuture = new CompletableFuture<>();
pendingFutures.add(newPendingFuture);
return newPendingFuture;
});
}
@Override
public E peek() {
final MeteringAgent.Context timer = monitor.startTimer(PEEK);
return Futures.getUnchecked(database.queuePeek(name)
.thenApply(v -> v != null ? serializer.<E>decode(v) : null)
.whenComplete((r, e) -> timer.stop(e)));
}
@Override
public String name() {
return name;
}
@Override
public DistributedPrimitive.Type primitiveType() {
return DistributedPrimitive.Type.QUEUE;
}
protected void tryPoll() {
Set<CompletableFuture<E>> completedFutures = Sets.newHashSet();
for (CompletableFuture<E> future : pendingFutures) {
E entry = Futures.getUnchecked(database.queuePop(name)
.thenApply(v -> v != null ? serializer.decode(v) : null));
if (entry != null) {
future.complete(entry);
completedFutures.add(future);
} else {
break;
}
}
pendingFutures.removeAll(completedFutures);
}
}

81
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueueBuilder.java
View File

@ -1,81 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import org.onosproject.store.service.DistributedQueue;
import org.onosproject.store.service.DistributedQueueBuilder;
import org.onosproject.store.service.Serializer;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
/**
* Default implementation of a {@code DistributedQueueBuilder}.
*
* @param <E> queue entry type
*/
public class DefaultDistributedQueueBuilder<E> implements DistributedQueueBuilder<E> {
private Serializer serializer;
private String name;
private boolean persistenceEnabled = true;
private final DatabaseManager databaseManager;
private boolean metering = true;
public DefaultDistributedQueueBuilder(DatabaseManager databaseManager) {
this.databaseManager = databaseManager;
}
@Override
public DistributedQueueBuilder<E> withName(String name) {
checkArgument(name != null && !name.isEmpty());
this.name = name;
return this;
}
@Override
public DistributedQueueBuilder<E> withSerializer(Serializer serializer) {
checkArgument(serializer != null);
this.serializer = serializer;
return this;
}
@Override
public DistributedQueueBuilder<E> withMeteringDisabled() {
metering = false;
return this;
}
@Override
public DistributedQueueBuilder<E> withPersistenceDisabled() {
persistenceEnabled = false;
return this;
}
private boolean validInputs() {
return name != null && serializer != null;
}
@Override
public DistributedQueue<E> build() {
checkState(validInputs());
return new DefaultDistributedQueue<>(
name,
persistenceEnabled ? databaseManager.partitionedDatabase : databaseManager.inMemoryDatabase,
serializer,
metering);
}
}

146
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java
View File

@ -1,146 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.function.Supplier;
import static com.google.common.base.Preconditions.*;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.primitives.resources.impl.CommitResult;
import org.onosproject.store.service.CommitStatus;
import org.onosproject.store.service.ConsistentMapBuilder;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.TransactionContext;
import org.onosproject.store.service.TransactionalMap;
import com.google.common.base.MoreObjects;
import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Futures;
/**
* Default TransactionContext implementation.
*/
public class DefaultTransactionContext implements TransactionContext {
private static final String TX_NOT_OPEN_ERROR = "Transaction Context is not open";
@SuppressWarnings("rawtypes")
private final Map<String, DefaultTransactionalMap> txMaps = Maps.newConcurrentMap();
private boolean isOpen = false;
private final Function<Transaction, CompletableFuture<CommitResult>> transactionCommitter;
private final TransactionId transactionId;
private final Supplier<ConsistentMapBuilder> mapBuilderSupplier;
public DefaultTransactionContext(TransactionId transactionId,
Function<Transaction, CompletableFuture<CommitResult>> transactionCommitter,
Supplier<ConsistentMapBuilder> mapBuilderSupplier) {
this.transactionId = transactionId;
this.transactionCommitter = checkNotNull(transactionCommitter);
this.mapBuilderSupplier = checkNotNull(mapBuilderSupplier);
}
@Override
public TransactionId transactionId() {
return transactionId;
}
@Override
public void begin() {
checkState(!isOpen, "Transaction Context is already open");
isOpen = true;
}
@Override
public boolean isOpen() {
return isOpen;
}
@Override
@SuppressWarnings("unchecked")
public <K, V> TransactionalMap<K, V> getTransactionalMap(String mapName,
Serializer serializer) {
checkState(isOpen, TX_NOT_OPEN_ERROR);
checkNotNull(mapName);
checkNotNull(serializer);
return txMaps.computeIfAbsent(mapName, name -> {
ConsistentMapBuilder mapBuilder = (ConsistentMapBuilder) mapBuilderSupplier.get()
.withName(name)
.withSerializer(serializer);
return new DefaultTransactionalMap<>(
name,
mapBuilder.buildAsyncMap(),
this,
serializer);
});
}
@SuppressWarnings("unchecked")
@Override
public CompletableFuture<CommitStatus> commit() {
// TODO: rework commit implementation to be more intuitive
checkState(isOpen, TX_NOT_OPEN_ERROR);
CommitStatus status;
try {
List<MapUpdate<String, byte[]>> updates = Lists.newLinkedList();
txMaps.values().forEach(m -> updates.addAll(m.toMapUpdates()));
Transaction transaction = new Transaction(transactionId, updates);
status = Futures.getUnchecked(transactionCommitter.apply(transaction)) == CommitResult.OK
? CommitStatus.SUCCESS : CommitStatus.FAILURE;
} catch (Exception e) {
abort();
status = CommitStatus.FAILURE;
} finally {
isOpen = false;
}
return CompletableFuture.completedFuture(status);
}
@Override
public void abort() {
if (isOpen) {
try {
txMaps.values().forEach(m -> m.abort());
} finally {
isOpen = false;
}
}
}
@Override
public String toString() {
ToStringHelper s = MoreObjects.toStringHelper(this)
.add("transactionId", transactionId)
.add("isOpen", isOpen);
txMaps.entrySet().forEach(e -> {
s.add(e.getKey(), e.getValue());
});
return s.toString();
}
@Override
public String name() {
return transactionId.toString();
}
}

56
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java
View File

@ -1,56 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.function.Supplier;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.primitives.resources.impl.CommitResult;
import org.onosproject.store.service.ConsistentMapBuilder;
import org.onosproject.store.service.TransactionContext;
import org.onosproject.store.service.TransactionContextBuilder;
/**
* The default implementation of a transaction context builder. This builder
* generates a {@link DefaultTransactionContext}.
*/
public class DefaultTransactionContextBuilder extends TransactionContextBuilder {
private final Supplier<ConsistentMapBuilder> mapBuilderSupplier;
private final Function<Transaction, CompletableFuture<CommitResult>> transactionCommitter;
private final TransactionId transactionId;
public DefaultTransactionContextBuilder(Supplier<ConsistentMapBuilder> mapBuilderSupplier,
Function<Transaction, CompletableFuture<CommitResult>> transactionCommiter,
TransactionId transactionId) {
this.mapBuilderSupplier = mapBuilderSupplier;
this.transactionCommitter = transactionCommiter;
this.transactionId = transactionId;
}
@Override
public TransactionContext build() {
return new DefaultTransactionContext(transactionId, transactionCommitter, () -> {
ConsistentMapBuilder mapBuilder = mapBuilderSupplier.get();
if (partitionsDisabled()) {
mapBuilder = (ConsistentMapBuilder) mapBuilder.withPartitionsDisabled();
}
return mapBuilder;
});
}
}

131
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MappingSet.java
View File

@ -1,131 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import com.google.common.collect.Iterators;
/**
* Set view backed by Set with element type {@code <BACK>} but returns
* element as {@code <OUT>} for convenience.
*
* @param <BACK> Backing {@link Set} element type.
* MappingSet will follow this type's equality behavior.
* @param <OUT> external facing element type.
* MappingSet will ignores equality defined by this type.
*/
class MappingSet<BACK, OUT> implements Set<OUT> {
private final Set<BACK> backedSet;
private final Function<OUT, BACK> toBack;
private final Function<BACK, OUT> toOut;
public MappingSet(Set<BACK> backedSet,
Function<Set<BACK>, Set<BACK>> supplier,
Function<OUT, BACK> toBack, Function<BACK, OUT> toOut) {
this.backedSet = supplier.apply(backedSet);
this.toBack = toBack;
this.toOut = toOut;
}
@Override
public int size() {
return backedSet.size();
}
@Override
public boolean isEmpty() {
return backedSet.isEmpty();
}
@Override
public boolean contains(Object o) {
return backedSet.contains(toBack.apply((OUT) o));
}
@Override
public Iterator<OUT> iterator() {
return Iterators.transform(backedSet.iterator(), toOut::apply);
}
@Override
public Object[] toArray() {
return backedSet.stream()
.map(toOut)
.toArray();
}
@Override
public <T> T[] toArray(T[] a) {
return backedSet.stream()
.map(toOut)
.toArray(size -> {
if (size < a.length) {
return (T[]) new Object[size];
} else {
Arrays.fill(a, null);
return a;
}
});
}
@Override
public boolean add(OUT e) {
return backedSet.add(toBack.apply(e));
}
@Override
public boolean remove(Object o) {
return backedSet.remove(toBack.apply((OUT) o));
}
@Override
public boolean containsAll(Collection<?> c) {
return c.stream()
.map(e -> toBack.apply((OUT) e))
.allMatch(backedSet::contains);
}
@Override
public boolean addAll(Collection<? extends OUT> c) {
return backedSet.addAll(c.stream().map(toBack).collect(Collectors.toList()));
}
@Override
public boolean retainAll(Collection<?> c) {
return backedSet.retainAll(c.stream()
.map(x -> toBack.apply((OUT) x))
.collect(Collectors.toList()));
}
@Override
public boolean removeAll(Collection<?> c) {
return backedSet.removeAll(c.stream()
.map(x -> toBack.apply((OUT) x))
.collect(Collectors.toList()));
}
@Override
public void clear() {
backedSet.clear();
}
}

404
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedDatabase.java
View File

@ -1,404 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import net.kuujo.copycat.Task;
import net.kuujo.copycat.cluster.Cluster;
import net.kuujo.copycat.resource.ResourceState;
import org.onlab.util.Match;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.resources.impl.CommitResult;
import org.onosproject.store.service.Versioned;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkState;
/**
* A database that partitions the keys across one or more database partitions.
*/
public class PartitionedDatabase implements Database {
private final String name;
private final Partitioner<String> partitioner;
private final List<Database> partitions;
private final AtomicBoolean isOpen = new AtomicBoolean(false);
private static final String DB_NOT_OPEN = "Partitioned Database is not open";
private TransactionManager transactionManager;
public PartitionedDatabase(
String name,
Collection<Database> partitions) {
this.name = name;
this.partitions = partitions
.stream()
.sorted((db1, db2) -> db1.name().compareTo(db2.name()))
.collect(Collectors.toList());
this.partitioner = new SimpleKeyHashPartitioner(this.partitions);
}
/**
* Returns the databases for individual partitions.
* @return list of database partitions
*/
public List<Database> getPartitions() {
return partitions;
}
/**
* Returns true if the database is open.
* @return true if open, false otherwise
*/
@Override
public boolean isOpen() {
return isOpen.get();
}
@Override
public CompletableFuture<Set<String>> maps() {
checkState(isOpen.get(), DB_NOT_OPEN);
Set<String> mapNames = Sets.newConcurrentHashSet();
return CompletableFuture.allOf(partitions
.stream()
.map(db -> db.maps().thenApply(mapNames::addAll))
.toArray(CompletableFuture[]::new))
.thenApply(v -> mapNames);
}
@Override
public CompletableFuture<Map<String, Long>> counters() {
checkState(isOpen.get(), DB_NOT_OPEN);
Map<String, Long> counters = Maps.newConcurrentMap();
return CompletableFuture.allOf(partitions
.stream()
.map(db -> db.counters()
.thenApply(m -> {
counters.putAll(m);
return null;
}))
.toArray(CompletableFuture[]::new))
.thenApply(v -> counters);
}
@Override
public CompletableFuture<Integer> mapSize(String mapName) {
checkState(isOpen.get(), DB_NOT_OPEN);
AtomicInteger totalSize = new AtomicInteger(0);
return CompletableFuture.allOf(partitions
.stream()
.map(p -> p.mapSize(mapName).thenApply(totalSize::addAndGet))
.toArray(CompletableFuture[]::new))
.thenApply(v -> totalSize.get());
}
@Override
public CompletableFuture<Boolean> mapIsEmpty(String mapName) {
checkState(isOpen.get(), DB_NOT_OPEN);
return mapSize(mapName).thenApply(size -> size == 0);
}
@Override
public CompletableFuture<Boolean> mapContainsKey(String mapName, String key) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(mapName, key).mapContainsKey(mapName, key);
}
@Override
public CompletableFuture<Boolean> mapContainsValue(String mapName, byte[] value) {
checkState(isOpen.get(), DB_NOT_OPEN);
AtomicBoolean containsValue = new AtomicBoolean(false);
return CompletableFuture.allOf(partitions
.stream()
.map(p -> p.mapContainsValue(mapName, value)
.thenApply(v -> containsValue.compareAndSet(false, v)))
.toArray(CompletableFuture[]::new))
.thenApply(v -> containsValue.get());
}
@Override
public CompletableFuture<Versioned<byte[]>> mapGet(String mapName, String key) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(mapName, key).mapGet(mapName, key);
}
@Override
public CompletableFuture<Result<UpdateResult<String, byte[]>>> mapUpdate(
String mapName, String key, Match<byte[]> valueMatch,
Match<Long> versionMatch, byte[] value) {
return partitioner.getPartition(mapName, key).mapUpdate(mapName, key, valueMatch, versionMatch, value);
}
@Override
public CompletableFuture<Result<Void>> mapClear(String mapName) {
AtomicBoolean isLocked = new AtomicBoolean(false);
checkState(isOpen.get(), DB_NOT_OPEN);
return CompletableFuture.allOf(partitions
.stream()
.map(p -> p.mapClear(mapName)
.thenApply(v -> isLocked.compareAndSet(false, Result.Status.LOCKED == v.status())))
.toArray(CompletableFuture[]::new))
.thenApply(v -> isLocked.get() ? Result.locked() : Result.ok(null));
}
@Override
public CompletableFuture<Set<String>> mapKeySet(String mapName) {
checkState(isOpen.get(), DB_NOT_OPEN);
Set<String> keySet = Sets.newConcurrentHashSet();
return CompletableFuture.allOf(partitions
.stream()
.map(p -> p.mapKeySet(mapName).thenApply(keySet::addAll))
.toArray(CompletableFuture[]::new))
.thenApply(v -> keySet);
}
@Override
public CompletableFuture<Collection<Versioned<byte[]>>> mapValues(String mapName) {
checkState(isOpen.get(), DB_NOT_OPEN);
List<Versioned<byte[]>> values = new CopyOnWriteArrayList<>();
return CompletableFuture.allOf(partitions
.stream()
.map(p -> p.mapValues(mapName).thenApply(values::addAll))
.toArray(CompletableFuture[]::new))
.thenApply(v -> values);
}
@Override
public CompletableFuture<Set<Entry<String, Versioned<byte[]>>>> mapEntrySet(String mapName) {
checkState(isOpen.get(), DB_NOT_OPEN);
Set<Entry<String, Versioned<byte[]>>> entrySet = Sets.newConcurrentHashSet();
return CompletableFuture.allOf(partitions
.stream()
.map(p -> p.mapEntrySet(mapName).thenApply(entrySet::addAll))
.toArray(CompletableFuture[]::new))
.thenApply(v -> entrySet);
}
@Override
public CompletableFuture<Long> counterGet(String counterName) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(counterName, counterName).counterGet(counterName);
}
@Override
public CompletableFuture<Long> counterAddAndGet(String counterName, long delta) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(counterName, counterName).counterAddAndGet(counterName, delta);
}
@Override
public CompletableFuture<Long> counterGetAndAdd(String counterName, long delta) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(counterName, counterName).counterGetAndAdd(counterName, delta);
}
@Override
public CompletableFuture<Void> counterSet(String counterName, long value) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(counterName, counterName).counterSet(counterName, value);
}
@Override
public CompletableFuture<Boolean> counterCompareAndSet(String counterName, long expectedValue, long updateValue) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(counterName, counterName).
counterCompareAndSet(counterName, expectedValue, updateValue);
}
@Override
public CompletableFuture<Long> queueSize(String queueName) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(queueName, queueName).queueSize(queueName);
}
@Override
public CompletableFuture<Void> queuePush(String queueName, byte[] entry) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(queueName, queueName).queuePush(queueName, entry);
}
@Override
public CompletableFuture<byte[]> queuePop(String queueName) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(queueName, queueName).queuePop(queueName);
}
@Override
public CompletableFuture<byte[]> queuePeek(String queueName) {
checkState(isOpen.get(), DB_NOT_OPEN);
return partitioner.getPartition(queueName, queueName).queuePeek(queueName);
}
@Override
public CompletableFuture<CommitResponse> prepareAndCommit(Transaction transaction) {
Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
if (subTransactions.isEmpty()) {
return CompletableFuture.completedFuture(CommitResponse.success(ImmutableList.of()));
} else if (subTransactions.size() == 1) {
Entry<Database, Transaction> entry =
subTransactions.entrySet().iterator().next();
return entry.getKey().prepareAndCommit(entry.getValue());
} else {
if (transactionManager == null) {
throw new IllegalStateException("TransactionManager is not initialized");
}
return transactionManager.execute(transaction)
.thenApply(r -> r == CommitResult.OK
? CommitResponse.success(ImmutableList.of()) : CommitResponse.failure());
}
}
@Override
public CompletableFuture<Boolean> prepare(Transaction transaction) {
Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
AtomicBoolean status = new AtomicBoolean(true);
return CompletableFuture.allOf(subTransactions.entrySet()
.stream()
.map(entry -> entry
.getKey()
.prepare(entry.getValue())
.thenApply(v -> status.compareAndSet(true, v)))
.toArray(CompletableFuture[]::new))
.thenApply(v -> status.get());
}
@Override
public CompletableFuture<CommitResponse> commit(Transaction transaction) {
Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
AtomicBoolean success = new AtomicBoolean(true);
List<UpdateResult<String, byte[]>> allUpdates = Lists.newArrayList();
return CompletableFuture.allOf(subTransactions.entrySet()
.stream()
.map(entry -> entry.getKey().commit(entry.getValue())
.thenAccept(response -> {
success.set(success.get() && response.success());
if (success.get()) {
allUpdates.addAll(response.updates());
}
}))
.toArray(CompletableFuture[]::new))
.thenApply(v -> success.get() ?
CommitResponse.success(allUpdates) : CommitResponse.failure());
}
@Override
public CompletableFuture<Boolean> rollback(Transaction transaction) {
Map<Database, Transaction> subTransactions = createSubTransactions(transaction);
return CompletableFuture.allOf(subTransactions.entrySet()
.stream()
.map(entry -> entry.getKey().rollback(entry.getValue()))
.toArray(CompletableFuture[]::new))
.thenApply(v -> true);
}
@Override
public CompletableFuture<Database> open() {
return CompletableFuture.allOf(partitions
.stream()
.map(Database::open)
.toArray(CompletableFuture[]::new))
.thenApply(v -> {
isOpen.set(true);
return this;
});
}
@Override
public CompletableFuture<Void> close() {
checkState(isOpen.get(), DB_NOT_OPEN);
return CompletableFuture.allOf(partitions
.stream()
.map(database -> database.close())
.toArray(CompletableFuture[]::new));
}
@Override
public boolean isClosed() {
return !isOpen.get();
}
@Override
public String name() {
return name;
}
@Override
public Cluster cluster() {
throw new UnsupportedOperationException();
}
@Override
public Database addStartupTask(Task<CompletableFuture<Void>> task) {
throw new UnsupportedOperationException();
}
@Override
public Database addShutdownTask(Task<CompletableFuture<Void>> task) {
throw new UnsupportedOperationException();
}
@Override
public ResourceState state() {
throw new UnsupportedOperationException();
}
private Map<Database, Transaction> createSubTransactions(
Transaction transaction) {
Map<Database, List<MapUpdate<String, byte[]>>> perPartitionUpdates = Maps.newHashMap();
for (MapUpdate<String, byte[]> update : transaction.updates()) {
Database partition = partitioner.getPartition(update.mapName(), update.key());
List<MapUpdate<String, byte[]>> partitionUpdates =
perPartitionUpdates.computeIfAbsent(partition, k -> Lists.newLinkedList());
partitionUpdates.add(update);
}
Map<Database, Transaction> subTransactions = Maps.newHashMap();
perPartitionUpdates.forEach((k, v) -> subTransactions.put(k, new Transaction(transaction.id(), v)));
return subTransactions;
}
protected void setTransactionManager(TransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
@Override
public void registerConsumer(Consumer<StateMachineUpdate> consumer) {
partitions.forEach(p -> p.registerConsumer(consumer));
}
@Override
public void unregisterConsumer(Consumer<StateMachineUpdate> consumer) {
partitions.forEach(p -> p.unregisterConsumer(consumer));
}
}

33
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Partitioner.java
View File

@ -1,33 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
/**
* Partitioner is responsible for mapping keys to individual database partitions.
*
* @param <K> key type.
*/
public interface Partitioner<K> {
/**
* Returns the database partition.
* @param mapName map name
* @param key key
* @return Database partition
*/
Database getPartition(String mapName, K key);
}

121
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Result.java
View File

@ -1,121 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static com.google.common.base.MoreObjects.toStringHelper;
import java.util.Objects;
/**
* Result of a database update operation.
*
* @param <V> return value type
*/
public final class Result<V> {
public enum Status {
/**
* Indicates a successful update.
*/
OK,
/**
* Indicates a failure due to underlying state being locked by another transaction.
*/
LOCKED
}
private final Status status;
private final V value;
/**
* Creates a new Result instance with the specified value with status set to Status.OK.
*
* @param <V> result value type
* @param value result value
* @return Result instance
*/
public static <V> Result<V> ok(V value) {
return new Result<>(value, Status.OK);
}
/**
* Creates a new Result instance with status set to Status.LOCKED.
*
* @param <V> result value type
* @return Result instance
*/
public static <V> Result<V> locked() {
return new Result<>(null, Status.LOCKED);
}
private Result(V value, Status status) {
this.value = value;
this.status = status;
}
/**
* Returns true if this result indicates a successful execution i.e status is Status.OK.
*
* @return true if successful, false otherwise
*/
public boolean success() {
return status == Status.OK;
}
/**
* Returns the status of database update operation.
*
* @return database update status
*/
public Status status() {
return status;
}
/**
* Returns the return value for the update.
*
* @return value returned by database update. If the status is another
* other than Status.OK, this returns a null
*/
public V value() {
return value;
}
@Override
public int hashCode() {
return Objects.hash(value, status);
}
@SuppressWarnings("unchecked")
@Override
public boolean equals(Object other) {
if (!(other instanceof Result)) {
return false;
}
Result<V> that = (Result<V>) other;
return Objects.equals(this.value, that.value) &&
Objects.equals(this.status, that.status);
}
@Override
public String toString() {
return toStringHelper(this)
.add("status", status)
.add("value", value)
.toString();
}
}

38
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleKeyHashPartitioner.java
View File

@ -1,38 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.List;
/**
* A simple Partitioner for mapping keys to database partitions.
* <p>
* This class uses a md5 hash based hashing scheme for hashing the key to
* a partition.
*
*/
public class SimpleKeyHashPartitioner extends DatabasePartitioner {
public SimpleKeyHashPartitioner(List<Database> partitions) {
super(partitions);
}
@Override
public Database getPartition(String mapName, String key) {
return partitions.get(hash(key) % partitions.size());
}
}

39
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleTableHashPartitioner.java
View File

@ -1,39 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.List;
/**
* A simple Partitioner that uses the map name hash to
* pick a partition.
* <p>
* This class uses a md5 hash based hashing scheme for hashing the map name to
* a partition. This partitioner maps all keys for a map to the same database
* partition.
*/
public class SimpleTableHashPartitioner extends DatabasePartitioner {
public SimpleTableHashPartitioner(List<Database> partitions) {
super(partitions);
}
@Override
public Database getPartition(String mapName, String key) {
return partitions.get(hash(mapName) % partitions.size());
}
}

91
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StateMachineUpdate.java
View File

@ -1,91 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static com.google.common.base.MoreObjects.toStringHelper;
/**
* Representation of a state machine update.
*/
public class StateMachineUpdate {
/**
* Target data structure type this update is for.
*/
enum Target {
/**
* Update is for a map.
*/
MAP_UPDATE,
/**
* Update is a transaction commit.
*/
TX_COMMIT,
/**
* Update is a queue push.
*/
QUEUE_PUSH,
/**
* Update is for some other operation.
*/
OTHER
}
private final String operationName;
private final Object input;
private final Object output;
public StateMachineUpdate(String operationName, Object input, Object output) {
this.operationName = operationName;
this.input = input;
this.output = output;
}
public Target target() {
// FIXME: This check is brittle
if (operationName.contains("mapUpdate")) {
return Target.MAP_UPDATE;
} else if (operationName.contains("commit") || operationName.contains("prepareAndCommit")) {
return Target.TX_COMMIT;
} else if (operationName.contains("queuePush")) {
return Target.QUEUE_PUSH;
} else {
return Target.OTHER;
}
}
@SuppressWarnings("unchecked")
public <T> T input() {
return (T) input;
}
@SuppressWarnings("unchecked")
public <T> T output() {
return (T) output;
}
@Override
public String toString() {
return toStringHelper(this)
.add("name", operationName)
.add("input", input)
.add("output", output)
.toString();
}
}

110
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java
View File

@ -1,110 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static com.google.common.base.Preconditions.checkNotNull;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.primitives.resources.impl.CommitResult;
import org.onosproject.store.service.AsyncConsistentMap;
import static org.onosproject.store.primitives.impl.Transaction.State.COMMITTED;
import static org.onosproject.store.primitives.impl.Transaction.State.COMMITTING;
import static org.onosproject.store.primitives.impl.Transaction.State.ROLLEDBACK;
import static org.onosproject.store.primitives.impl.Transaction.State.ROLLINGBACK;
/**
* Agent that runs the two phase commit protocol.
*/
public class TransactionManager {
private final Database database;
private final AsyncConsistentMap<TransactionId, Transaction> transactions;
public TransactionManager(Database database, AsyncConsistentMap<TransactionId, Transaction> transactions) {
this.database = checkNotNull(database, "database cannot be null");
this.transactions = transactions;
}
/**
* Executes the specified transaction by employing a two phase commit protocol.
*
* @param transaction transaction to commit
* @return transaction commit result
*/
public CompletableFuture<CommitResult> execute(Transaction transaction) {
// short-circuit if there is only a single update
if (transaction.updates().size() <= 1) {
return database.prepareAndCommit(transaction)
.thenApply(response -> response.success()
? CommitResult.OK : CommitResult.FAILURE_DURING_COMMIT);
}
// clean up if this transaction in already in a terminal state.
if (transaction.state() == COMMITTED || transaction.state() == ROLLEDBACK) {
return transactions.remove(transaction.id()).thenApply(v -> CommitResult.OK);
} else if (transaction.state() == COMMITTING) {
return commit(transaction);
} else if (transaction.state() == ROLLINGBACK) {
return rollback(transaction).thenApply(v -> CommitResult.FAILURE_TO_PREPARE);
} else {
return prepare(transaction).thenCompose(v -> v ? commit(transaction) : rollback(transaction));
}
}
/**
* Returns all pending transaction identifiers.
*
* @return future for a collection of transaction identifiers.
*/
public CompletableFuture<Collection<TransactionId>> getPendingTransactionIds() {
return transactions.values().thenApply(c -> c.stream()
.map(v -> v.value())
.filter(v -> v.state() != COMMITTED && v.state() != ROLLEDBACK)
.map(Transaction::id)
.collect(Collectors.toList()));
}
private CompletableFuture<Boolean> prepare(Transaction transaction) {
return transactions.put(transaction.id(), transaction)
.thenCompose(v -> database.prepare(transaction))
.thenCompose(status -> transactions.put(
transaction.id(),
transaction.transition(status ? COMMITTING : ROLLINGBACK))
.thenApply(v -> status));
}
private CompletableFuture<CommitResult> commit(Transaction transaction) {
return database.commit(transaction)
.thenCompose(r -> {
if (r.success()) {
return transactions.put(transaction.id(), transaction.transition(COMMITTED))
.thenApply(v -> CommitResult.OK);
} else {
return CompletableFuture.completedFuture(CommitResult.FAILURE_DURING_COMMIT);
}
});
}
private CompletableFuture<CommitResult> rollback(Transaction transaction) {
return database.rollback(transaction)
.thenCompose(v -> transactions.put(transaction.id(), transaction.transition(ROLLEDBACK)))
.thenApply(v -> CommitResult.FAILURE_TO_PREPARE);
}
}

82
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateResult.java
View File

@ -1,82 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import java.util.function.Function;
import org.onosproject.store.service.MapEvent;
import org.onosproject.store.service.Versioned;
/**
* Result of a update operation.
* <p>
* Both old and new values are accessible along with a flag that indicates if the
* the value was updated. If flag is false, oldValue and newValue both
* point to the same unmodified value.
* @param <V> result type
*/
public class UpdateResult<K, V> {
private final boolean updated;
private final String mapName;
private final K key;
private final Versioned<V> oldValue;
private final Versioned<V> newValue;
public UpdateResult(boolean updated, String mapName, K key, Versioned<V> oldValue, Versioned<V> newValue) {
this.updated = updated;
this.mapName = mapName;
this.key = key;
this.oldValue = oldValue;
this.newValue = newValue;
}
public boolean updated() {
return updated;
}
public String mapName() {
return mapName;
}
public K key() {
return key;
}
public Versioned<V> oldValue() {
return oldValue;
}
public Versioned<V> newValue() {
return newValue;
}
public <K1, V1> UpdateResult<K1, V1> map(Function<K, K1> keyTransform, Function<V, V1> valueMapper) {
return new UpdateResult<>(updated,
mapName,
keyTransform.apply(key),
oldValue == null ? null : oldValue.map(valueMapper),
newValue == null ? null : newValue.map(valueMapper));
}
public MapEvent<K, V> toMapEvent() {
if (!updated) {
return null;
} else {
return new MapEvent<>(mapName(), key(), newValue, oldValue);
}
}
}

373
core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/DefaultAsyncConsistentMapTest.java
View File

@ -1,373 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static java.util.Collections.unmodifiableCollection;
import static java.util.Collections.unmodifiableSet;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import java.util.Collection;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import net.kuujo.copycat.Task;
import net.kuujo.copycat.cluster.Cluster;
import net.kuujo.copycat.resource.ResourceState;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.onlab.util.Match;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.DefaultApplicationId;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.Versioned;
import com.google.common.base.MoreObjects;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
/**
*
*/
public class DefaultAsyncConsistentMapTest {
private static final ApplicationId APP_ID = new DefaultApplicationId(42, "what");
private static final TestData KEY1A = new TestData("One", "a");
private static final TestData KEY1B = new TestData("One", "b");
private static final TestData VALUE2A = new TestData("Two", "a");
private static final TestData VALUE2B = new TestData("Two", "b");
@Before
public void setUp() throws Exception {
}
@After
public void tearDown() throws Exception {
}
@Test
public void testKeySet() throws Exception {
DefaultAsyncConsistentMap<TestData, TestData> map;
String name = "map_name";
Database database = new TestDatabase();
Serializer serializer = Serializer.forTypes(TestData.class);
map = new DefaultAsyncConsistentMap<>(name, APP_ID, database, serializer,
false, false, false);
map.put(KEY1A, VALUE2A);
map.put(KEY1B, VALUE2A);
Set<TestData> set = map.keySet().get();
assertEquals("Should contain 2 keys",
2, set.size());
assertThat(set.contains(KEY1A), is(true));
assertThat(set.contains(KEY1B), is(true));
assertThat(set.contains(new TestData("One", "a")), is(true));
}
@Test
public void testEntrySet() throws Exception {
DefaultAsyncConsistentMap<TestData, TestData> map;
String name = "map_name";
Database database = new TestDatabase();
Serializer serializer = Serializer.forTypes(TestData.class);
map = new DefaultAsyncConsistentMap<>(name, APP_ID, database, serializer,
false, false, false);
map.put(KEY1A, VALUE2A);
map.put(KEY1B, VALUE2A);
assertEquals("Should contain 2 entry",
2,
map.entrySet().get().size());
}
/**
* Object to be used as a test data.
*
* {@link Object#equals(Object)} use only part of it's fields.
*
* As a result there can be 2 instances which the
* serialized bytes are not-equal but
* {@link Object#equals(Object)}-wise they are equal.
*/
public static class TestData {
private final String theKey;
@SuppressWarnings("unused")
private final String notUsedForEquals;
public TestData(String theKey, String notUsedForEquals) {
this.theKey = theKey;
this.notUsedForEquals = notUsedForEquals;
}
@Override
public int hashCode() {
return Objects.hashCode(theKey);
}
@Override
public boolean equals(Object obj) {
if (obj instanceof TestData) {
TestData that = (TestData) obj;
return Objects.equals(this.theKey, that.theKey);
}
return false;
}
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("theKey", theKey)
.add("notUsedForEquals", notUsedForEquals)
.toString();
}
}
/**
* {@link Database} implementation for testing.
*
* There is only 1 backing Map, {@code mapName} will be ignored.
*/
public class TestDatabase implements Database {
Map<String, Versioned<byte[]>> map = new ConcurrentHashMap<>();
@Override
public CompletableFuture<Set<String>> maps() {
return CompletableFuture.completedFuture(ImmutableSet.of());
}
@Override
public CompletableFuture<Map<String, Long>> counters() {
return CompletableFuture.completedFuture(ImmutableMap.of());
}
@Override
public CompletableFuture<Integer> mapSize(String mapName) {
return CompletableFuture.completedFuture(map.size());
}
@Override
public CompletableFuture<Boolean> mapIsEmpty(String mapName) {
return CompletableFuture.completedFuture(map.isEmpty());
}
@Override
public CompletableFuture<Boolean> mapContainsKey(String mapName,
String key) {
return CompletableFuture.completedFuture(map.containsKey(key));
}
@Override
public CompletableFuture<Boolean> mapContainsValue(String mapName,
byte[] value) {
return CompletableFuture.completedFuture(Maps.transformValues(map, Versioned::value)
.containsValue(value));
}
@Override
public CompletableFuture<Versioned<byte[]>> mapGet(String mapName,
String key) {
return CompletableFuture.completedFuture(map.get(key));
}
@Override
public synchronized CompletableFuture<Result<UpdateResult<String, byte[]>>> mapUpdate(String mapName,
String key,
Match<byte[]> valueMatch,
Match<Long> versionMatch,
byte[] value) {
boolean updated = false;
final Versioned<byte[]> oldValue;
final Versioned<byte[]> newValue;
Versioned<byte[]> old = map.getOrDefault(key, new Versioned<byte[]>(null, 0));
if (valueMatch.matches(old.value()) && versionMatch.matches(old.version())) {
updated = true;
oldValue = old;
newValue = new Versioned<>(value, old.version() + 1);
map.put(key, newValue);
} else {
updated = false;
oldValue = old;
newValue = old;
}
return CompletableFuture.completedFuture(
Result.ok(new UpdateResult<String, byte[]>(updated,
mapName, key, oldValue, newValue)));
}
@Override
public CompletableFuture<Result<Void>> mapClear(String mapName) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Set<String>> mapKeySet(String mapName) {
return CompletableFuture.completedFuture(unmodifiableSet(map.keySet()));
}
@Override
public CompletableFuture<Collection<Versioned<byte[]>>> mapValues(String mapName) {
return CompletableFuture.completedFuture(unmodifiableCollection(map.values()));
}
@Override
public CompletableFuture<Set<Entry<String, Versioned<byte[]>>>> mapEntrySet(String mapName) {
return CompletableFuture.completedFuture(unmodifiableSet(map.entrySet()));
}
@Override
public CompletableFuture<Long> counterAddAndGet(String counterName,
long delta) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Long> counterGetAndAdd(String counterName,
long delta) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Void> counterSet(String counterName,
long value) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Boolean> counterCompareAndSet(String counterName,
long expectedValue,
long update) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Long> counterGet(String counterName) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Long> queueSize(String queueName) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Void> queuePush(String queueName,
byte[] entry) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<byte[]> queuePop(String queueName) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<byte[]> queuePeek(String queueName) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<CommitResponse> prepareAndCommit(Transaction transaction) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Boolean> prepare(Transaction transaction) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<CommitResponse> commit(Transaction transaction) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Boolean> rollback(Transaction transaction) {
throw new UnsupportedOperationException();
}
@Override
public String name() {
return "name";
}
@Override
public ResourceState state() {
return ResourceState.HEALTHY;
}
@Override
public Cluster cluster() {
throw new UnsupportedOperationException();
}
@Override
public Database addStartupTask(Task<CompletableFuture<Void>> task) {
throw new UnsupportedOperationException();
}
@Override
public Database addShutdownTask(Task<CompletableFuture<Void>> task) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<Database> open() {
return CompletableFuture.completedFuture(this);
}
@Override
public boolean isOpen() {
return true;
}
@Override
public CompletableFuture<Void> close() {
return CompletableFuture.completedFuture(null);
}
@Override
public boolean isClosed() {
return false;
}
@Override
public void registerConsumer(Consumer<StateMachineUpdate> consumer) {
}
@Override
public void unregisterConsumer(Consumer<StateMachineUpdate> consumer) {
}
}
}

57
core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/ResultTest.java
View File

@ -1,57 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static junit.framework.TestCase.assertEquals;
import static junit.framework.TestCase.assertFalse;
import static junit.framework.TestCase.assertNull;
import static junit.framework.TestCase.assertTrue;
import org.junit.Test;
/**
* Unit tests for Result.
*/
public class ResultTest {
@Test
public void testLocked() {
Result<String> r = Result.locked();
assertFalse(r.success());
assertNull(r.value());
assertEquals(Result.Status.LOCKED, r.status());
}
@Test
public void testOk() {
Result<String> r = Result.ok("foo");
assertTrue(r.success());
assertEquals("foo", r.value());
assertEquals(Result.Status.OK, r.status());
}
@Test
public void testEquality() {
Result<String> r1 = Result.ok("foo");
Result<String> r2 = Result.locked();
Result<String> r3 = Result.ok("bar");
Result<String> r4 = Result.ok("foo");
assertTrue(r1.equals(r4));
assertFalse(r1.equals(r2));
assertFalse(r1.equals(r3));
assertFalse(r2.equals(r3));
}
}

99
core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/UpdateResultTest.java
View File

@ -1,99 +0,0 @@
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.store.primitives.impl;
import static junit.framework.TestCase.assertEquals;
import static junit.framework.TestCase.assertNull;
import static junit.framework.TestCase.assertTrue;
import org.junit.Test;
import org.onosproject.store.service.MapEvent;
import org.onosproject.store.service.Versioned;
/**
* Unit tests for UpdateResult.
*/
public class UpdateResultTest {
@Test
public void testGetters() {
Versioned<String> oldValue = new Versioned<>("a", 1);
Versioned<String> newValue = new Versioned<>("b", 2);
UpdateResult<String, String> ur =
new UpdateResult<>(true, "foo", "k", oldValue, newValue);
assertTrue(ur.updated());
assertEquals("foo", ur.mapName());
assertEquals("k", ur.key());
assertEquals(oldValue, ur.oldValue());
assertEquals(newValue, ur.newValue());
}
@Test
public void testToMapEvent() {
Versioned<String> oldValue = new Versioned<>("a", 1);
Versioned<String> newValue = new Versioned<>("b", 2);
UpdateResult<String, String> ur1 =
new UpdateResult<>(true, "foo", "k", oldValue, newValue);
MapEvent<String, String> event1 = ur1.toMapEvent();
assertEquals(MapEvent.Type.UPDATE, event1.type());
assertEquals("k", event1.key());
assertEquals(newValue, event1.value());
UpdateResult<String, String> ur2 =
new UpdateResult<>(true, "foo", "k", null, newValue);
MapEvent<String, String> event2 = ur2.toMapEvent();
assertEquals(MapEvent.Type.INSERT, event2.type());
assertEquals("k", event2.key());
assertEquals(newValue, event2.value());
UpdateResult<String, String> ur3 =
new UpdateResult<>(true, "foo", "k", oldValue, null);
MapEvent<String, String> event3 = ur3.toMapEvent();
assertEquals(MapEvent.Type.REMOVE, event3.type());
assertEquals("k", event3.key());
assertEquals(oldValue, event3.value());
UpdateResult<String, String> ur4 =
new UpdateResult<>(false, "foo", "k", oldValue, oldValue);
assertNull(ur4.toMapEvent());
}
@Test
public void testMap() {
Versioned<String> oldValue = new Versioned<>("a", 1);
Versioned<String> newValue = new Versioned<>("b", 2);
UpdateResult<String, String> ur1 =
new UpdateResult<>(true, "foo", "k", oldValue, newValue);
UpdateResult<Integer, Integer> ur2 = ur1.map(s -> s.length(), s -> s.length());
assertEquals(ur2.updated(), ur1.updated());
assertEquals(ur1.mapName(), ur2.mapName());
assertEquals(new Integer(1), ur2.key());
assertEquals(oldValue.map(s -> s.length()), ur2.oldValue());
assertEquals(newValue.map(s -> s.length()), ur2.newValue());
UpdateResult<String, String> ur3 =
new UpdateResult<>(true, "foo", "k", null, newValue);
UpdateResult<Integer, Integer> ur4 = ur3.map(s -> s.length(), s -> s.length());
assertEquals(ur3.updated(), ur4.updated());
assertEquals(ur3.mapName(), ur4.mapName());
assertEquals(new Integer(1), ur4.key());
assertNull(ur4.oldValue());
assertEquals(newValue.map(s -> s.length()), ur4.newValue());
}
}

3
pom.xml
View File

@ -80,8 +80,7 @@
<!-- TODO: replace with final release version when it is out -->
<catalyst.version>1.0.6</catalyst.version>
<atomix.version>1.0.0-rc3</atomix.version>
<atomix.copycat.version>1.0.0-rc6</atomix.copycat.version>
<copycat.version>0.5.1.onos</copycat.version>
<copycat.version>1.0.0-rc6</copycat.version>
<openflowj.version>0.9.3.onos-SNAPSHOT</openflowj.version>
<onos-maven-plugin.version>1.9</onos-maven-plugin.version>
<osgi.version>4.3.1</osgi.version>

27
utils/thirdparty/pom.xml vendored
View File

@ -53,28 +53,15 @@
<dependency>
<groupId>io.atomix.copycat</groupId>
<artifactId>copycat-client</artifactId>
<version>${atomix.copycat.version}</version>
<version>${copycat.version}</version>
</dependency>
<dependency>
<groupId>io.atomix.copycat</groupId>
<artifactId>copycat-server</artifactId>
<version>${atomix.copycat.version}</version>
</dependency>
<dependency>
<!-- FIXME once fixes get merged to upstream -->
<groupId>org.onosproject</groupId>
<artifactId>copycat-api</artifactId>
<version>${copycat.version}</version>
</dependency>
<dependency>
<!-- FIXME once fixes get merged to upstream -->
<groupId>org.onosproject</groupId>
<artifactId>copycat-core</artifactId>
<version>${copycat.version}</version>
</dependency>
</dependencies>
<build>
@ -98,16 +85,6 @@
</artifactSet>
<filters>
<filter>
<artifact>org.onosproject:copycat*</artifact>
<includes>
<include>**</include>
</includes>
<excludes>
<exclude>net/kuujo/copycat/**</exclude>
</excludes>
</filter>
<filter>
<artifact>io.atomix:atomix-all</artifact>
<includes>
@ -132,7 +109,7 @@
<configuration>
<instructions>
<Export-Package>
net.kuujo.copycat.*;io.atomix.*
io.atomix.*
</Export-Package>
<Import-Package>
!sun.nio.ch,!sun.misc,*