Refactor change: Moved all distributed primitive implementation classes into onos-core-primitives bundle
Change-Id: Icd5dbd4133cb2f21bd403bcd598e6012813e6bfd
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AntiEntropyAdvertisement.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AntiEntropyAdvertisement.java
new file mode 100644
index 0000000..e7143b6
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AntiEntropyAdvertisement.java
@@ -0,0 +1,71 @@
+/*
+ * 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.MoreObjects;
+import com.google.common.collect.ImmutableMap;
+
+import org.onosproject.cluster.NodeId;
+
+import java.util.Map;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Anti-entropy advertisement message for eventually consistent map.
+ */
+public class AntiEntropyAdvertisement<K> {
+
+ private final NodeId sender;
+ private final Map<K, MapValue.Digest> digest;
+
+ /**
+ * Creates a new anti entropy advertisement message.
+ *
+ * @param sender the sender's node ID
+ * @param digest for map entries
+ */
+ public AntiEntropyAdvertisement(NodeId sender,
+ Map<K, MapValue.Digest> digest) {
+ this.sender = checkNotNull(sender);
+ this.digest = ImmutableMap.copyOf(checkNotNull(digest));
+ }
+
+ /**
+ * Returns the sender's node ID.
+ *
+ * @return the sender's node ID
+ */
+ public NodeId sender() {
+ return sender;
+ }
+
+ /**
+ * Returns the digest for map entries.
+ *
+ * @return mapping from key to associated digest
+ */
+ public Map<K, MapValue.Digest> digest() {
+ return digest;
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("sender", sender)
+ .add("totalEntries", digest.size())
+ .toString();
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AsyncCachingConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AsyncCachingConsistentMap.java
new file mode 100644
index 0000000..9df1b3b
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/AsyncCachingConsistentMap.java
@@ -0,0 +1,81 @@
+/*
+ * 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);
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CommitResponse.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CommitResponse.java
new file mode 100644
index 0000000..a18ade4
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CommitResponse.java
@@ -0,0 +1,61 @@
+/*
+ * 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();
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ConsistentMapBackedJavaMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ConsistentMapBackedJavaMap.java
new file mode 100644
index 0000000..7c23c65
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ConsistentMapBackedJavaMap.java
@@ -0,0 +1,160 @@
+/*
+ * 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.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.Versioned;
+
+import com.google.common.collect.Collections2;
+import com.google.common.collect.Maps;
+
+/**
+ * Standard java Map backed by a ConsistentMap.
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public final class ConsistentMapBackedJavaMap<K, V> implements Map<K, V> {
+
+ private final ConsistentMap<K, V> backingMap;
+
+ public ConsistentMapBackedJavaMap(ConsistentMap<K, V> backingMap) {
+ this.backingMap = backingMap;
+ }
+
+ @Override
+ public int size() {
+ return backingMap.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return backingMap.isEmpty();
+ }
+
+ @Override
+ public boolean containsKey(Object key) {
+ return backingMap.containsKey((K) key);
+ }
+
+ @Override
+ public boolean containsValue(Object value) {
+ return backingMap.containsValue((V) value);
+ }
+
+ @Override
+ public V get(Object key) {
+ return Versioned.valueOrElse(backingMap.get((K) key), null);
+ }
+
+ @Override
+ public V getOrDefault(Object key, V defaultValue) {
+ return Versioned.valueOrElse(backingMap.get((K) key), defaultValue);
+ }
+
+ @Override
+ public V put(K key, V value) {
+ return Versioned.valueOrElse(backingMap.put(key, value), null);
+ }
+
+ @Override
+ public V putIfAbsent(K key, V value) {
+ return Versioned.valueOrElse(backingMap.putIfAbsent(key, value), null);
+ }
+
+ @Override
+ public V remove(Object key) {
+ return Versioned.valueOrElse(backingMap.remove((K) key), null);
+ }
+
+ @Override
+ public boolean remove(Object key, Object value) {
+ return backingMap.remove((K) key, (V) value);
+ }
+
+ @Override
+ public V replace(K key, V value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean replace(K key, V oldValue, V newValue) {
+ return backingMap.replace(key, oldValue, newValue);
+ }
+
+ @Override
+ public void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach((k, v) -> {
+ backingMap.put(k, v);
+ });
+ }
+
+ @Override
+ public void clear() {
+ backingMap.clear();
+ }
+
+ @Override
+ public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+ return Versioned.valueOrElse(backingMap.compute(key, remappingFunction), null);
+ }
+
+ @Override
+ public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
+ return Versioned.valueOrElse(backingMap.computeIfAbsent(key, mappingFunction), null);
+ }
+
+ @Override
+ public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+ return Versioned.valueOrElse(backingMap.computeIfPresent(key, remappingFunction), null);
+ }
+
+ @Override
+ public Set<K> keySet() {
+ return backingMap.keySet();
+ }
+
+ @Override
+ public Collection<V> values() {
+ return Collections2.transform(backingMap.values(), v -> v.value());
+ }
+
+ @Override
+ public Set<java.util.Map.Entry<K, V>> entrySet() {
+ return backingMap.entrySet()
+ .stream()
+ .map(entry -> Maps.immutableEntry(entry.getKey(), entry.getValue().value()))
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public void forEach(BiConsumer<? super K, ? super V> action) {
+ entrySet().forEach(e -> action.accept(e.getKey(), e.getValue()));
+ }
+
+ @Override
+ public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
+ return computeIfPresent(key, (k, v) -> v == null ? value : remappingFunction.apply(v, value));
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatCommunicationProtocol.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatCommunicationProtocol.java
new file mode 100644
index 0000000..e40665f
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatCommunicationProtocol.java
@@ -0,0 +1,134 @@
+/*
+ * 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);
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Database.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Database.java
new file mode 100644
index 0000000..bacf9f6
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Database.java
@@ -0,0 +1,106 @@
+/*
+ * 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);
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseConfig.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseConfig.java
new file mode 100644
index 0000000..57dd31c
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseConfig.java
@@ -0,0 +1,157 @@
+/*
+ * 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);
+ }
+
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseManager.java
new file mode 100644
index 0000000..b35b930
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseManager.java
@@ -0,0 +1,433 @@
+/*
+ * 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.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.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.commons.lang.math.RandomUtils;
+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.core.IdGenerator;
+import org.onosproject.persistence.PersistenceService;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+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.MapInfo;
+import org.onosproject.store.service.PartitionInfo;
+import org.onosproject.store.service.StorageAdminService;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Transaction;
+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 = true)
+@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 IdGenerator transactionIdGenerator = () -> RandomUtils.nextLong();
+
+ 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.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);
+
+ transactionManager = new TransactionManager(partitionedDatabase, consistentMapBuilder());
+ 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, transactionIdGenerator.getNewId());
+ }
+
+ @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);
+ }
+
+
+ @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() {
+ return new DefaultAtomicValueBuilder<>(this);
+ }
+
+ @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<Transaction> getTransactions() {
+ return complete(transactionManager.getTransactions());
+ }
+
+ 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());
+ }
+ }
+
+ @Override
+ public void redriveTransactions() {
+ getTransactions().stream().forEach(transactionManager::execute);
+ }
+
+ 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());
+ }
+ }
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabasePartitioner.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabasePartitioner.java
new file mode 100644
index 0000000..f5ccdda
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabasePartitioner.java
@@ -0,0 +1,45 @@
+/*
+ * 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());
+ }
+
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseProxy.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseProxy.java
new file mode 100644
index 0000000..5c675eb
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseProxy.java
@@ -0,0 +1,250 @@
+/*
+ * 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.Transaction;
+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);
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseSerializer.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseSerializer.java
new file mode 100644
index 0000000..872bbeb
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseSerializer.java
@@ -0,0 +1,104 @@
+/*
+ * 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.serializers.KryoNamespaces;
+import org.onosproject.store.serializers.KryoSerializer;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
+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(DatabaseUpdate.class)
+ .register(DatabaseUpdate.Type.class)
+ .register(Result.class)
+ .register(UpdateResult.class)
+ .register(Result.Status.class)
+ .register(DefaultTransaction.class)
+ .register(Transaction.State.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);
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java
new file mode 100644
index 0000000..de308f5
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java
@@ -0,0 +1,118 @@
+/*
+ * 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.Transaction;
+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
+ CommitResponse prepareAndCommit(Transaction transaction);
+
+ @Command
+ boolean prepare(Transaction transaction);
+
+ @Command
+ CommitResponse commit(Transaction transaction);
+
+ @Command
+ boolean rollback(Transaction transaction);
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicCounter.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicCounter.java
new file mode 100644
index 0000000..1a38727
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicCounter.java
@@ -0,0 +1,106 @@
+/*
+ * 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 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));
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicValue.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicValue.java
new file mode 100644
index 0000000..912d6d0
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncAtomicValue.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2016 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.AsyncAtomicValue;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.AtomicValueEvent;
+import org.onosproject.store.service.AtomicValueEventListener;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Versioned;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+/**
+ * Default implementation of {@link AsyncAtomicValue}.
+ *
+ * @param <V> value type
+ */
+public class DefaultAsyncAtomicValue<V> implements AsyncAtomicValue<V> {
+
+ private final Set<AtomicValueEventListener<V>> listeners = new CopyOnWriteArraySet<>();
+ private final AsyncConsistentMap<String, V> valueMap;
+ private final String name;
+ private final MapEventListener<String, V> mapEventListener = new InternalMapEventListener();
+ private final MeteringAgent monitor;
+
+ private static final String COMPONENT_NAME = "atomicValue";
+ private static final String GET = "get";
+ private static final String GET_AND_SET = "getAndSet";
+ private static final String SET = "set";
+ private static final String COMPARE_AND_SET = "compareAndSet";
+
+ public DefaultAsyncAtomicValue(AsyncConsistentMap<String, V> valueMap,
+ String name,
+ boolean meteringEnabled) {
+ this.valueMap = valueMap;
+ this.name = name;
+ this.monitor = new MeteringAgent(COMPONENT_NAME, name, meteringEnabled);
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public CompletableFuture<Boolean> compareAndSet(V expect, V update) {
+ final MeteringAgent.Context newTimer = monitor.startTimer(COMPARE_AND_SET);
+ CompletableFuture<Boolean> response;
+ if (expect == null) {
+ if (update == null) {
+ response = CompletableFuture.completedFuture(true);
+ }
+ response = valueMap.putIfAbsent(name, update).thenApply(v -> v == null);
+ } else {
+ response = update == null
+ ? valueMap.remove(name, expect)
+ : valueMap.replace(name, expect, update);
+ }
+ return response.whenComplete((r, e) -> newTimer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<V> get() {
+ final MeteringAgent.Context newTimer = monitor.startTimer(GET);
+ return valueMap.get(name)
+ .thenApply(Versioned::valueOrNull)
+ .whenComplete((r, e) -> newTimer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<V> getAndSet(V value) {
+ final MeteringAgent.Context newTimer = monitor.startTimer(GET_AND_SET);
+ CompletableFuture<Versioned<V>> previousValue = value == null ?
+ valueMap.remove(name) : valueMap.put(name, value);
+ return previousValue.thenApply(Versioned::valueOrNull)
+ .whenComplete((r, e) -> newTimer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Void> set(V value) {
+ final MeteringAgent.Context newTimer = monitor.startTimer(SET);
+ CompletableFuture<Void> previousValue = value == null ?
+ valueMap.remove(name).thenApply(v -> null) : valueMap.put(name, value).thenApply(v -> null);
+ return previousValue.whenComplete((r, e) -> newTimer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Void> addListener(AtomicValueEventListener<V> listener) {
+ synchronized (listeners) {
+ if (listeners.add(listener)) {
+ if (listeners.size() == 1) {
+ return valueMap.addListener(mapEventListener);
+ }
+ }
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> removeListener(AtomicValueEventListener<V> listener) {
+ synchronized (listeners) {
+ if (listeners.remove(listener)) {
+ if (listeners.size() == 0) {
+ return valueMap.removeListener(mapEventListener);
+ }
+ }
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ private class InternalMapEventListener implements MapEventListener<String, V> {
+
+ @Override
+ public void event(MapEvent<String, V> mapEvent) {
+ V newValue = mapEvent.type() == MapEvent.Type.REMOVE ? null : mapEvent.value().value();
+ AtomicValueEvent<V> atomicValueEvent = new AtomicValueEvent<>(name, AtomicValueEvent.Type.UPDATE, newValue);
+ listeners.forEach(l -> l.event(atomicValueEvent));
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncConsistentMap.java
new file mode 100644
index 0000000..2ee93ae
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncConsistentMap.java
@@ -0,0 +1,505 @@
+/*
+ * 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.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.Maps;
+
+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.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.Serializer;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+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 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;
+
+/**
+ * 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);
+ }
+
+ 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);
+ }
+ });
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncDistributedSet.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncDistributedSet.java
new file mode 100644
index 0000000..aaa44aa
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAsyncDistributedSet.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2015-2016 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.Collection;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import org.onlab.util.Tools;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.AsyncDistributedSet;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.SetEvent;
+import org.onosproject.store.service.SetEventListener;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+/**
+ * Implementation of {@link AsyncDistributedSet}.
+ *
+ * @param <E> set entry type
+ */
+public class DefaultAsyncDistributedSet<E> implements AsyncDistributedSet<E> {
+
+ private static final String CONTAINS = "contains";
+ private static final String PRIMITIVE_NAME = "distributedSet";
+ private static final String SIZE = "size";
+ private static final String IS_EMPTY = "isEmpty";
+ private static final String ADD = "add";
+ private static final String REMOVE = "remove";
+ private static final String CONTAINS_ALL = "containsAll";
+ private static final String ADD_ALL = "addAll";
+ private static final String RETAIN_ALL = "retainAll";
+ private static final String REMOVE_ALL = "removeAll";
+ private static final String CLEAR = "clear";
+ private static final String GET_AS_IMMUTABLE_SET = "getAsImmutableSet";
+
+ private final String name;
+ private final AsyncConsistentMap<E, Boolean> backingMap;
+ private final Map<SetEventListener<E>, MapEventListener<E, Boolean>> listenerMapping = Maps.newIdentityHashMap();
+ private final MeteringAgent monitor;
+
+ public DefaultAsyncDistributedSet(AsyncConsistentMap<E, Boolean> backingMap, String name, boolean meteringEnabled) {
+ this.backingMap = backingMap;
+ this.name = name;
+ monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public CompletableFuture<Integer> size() {
+ final MeteringAgent.Context timer = monitor.startTimer(SIZE);
+ return backingMap.size().whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> isEmpty() {
+ final MeteringAgent.Context timer = monitor.startTimer(IS_EMPTY);
+ return backingMap.isEmpty().whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> contains(E element) {
+ final MeteringAgent.Context timer = monitor.startTimer(CONTAINS);
+ return backingMap.containsKey(element).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> add(E entry) {
+ final MeteringAgent.Context timer = monitor.startTimer(ADD);
+ return backingMap.putIfAbsent(entry, true).thenApply(Objects::isNull).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> remove(E entry) {
+ final MeteringAgent.Context timer = monitor.startTimer(REMOVE);
+ return backingMap.remove(entry, true).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> containsAll(Collection<? extends E> c) {
+ final MeteringAgent.Context timer = monitor.startTimer(CONTAINS_ALL);
+ return Tools.allOf(c.stream().map(this::contains).collect(Collectors.toList())).thenApply(v ->
+ v.stream().reduce(Boolean::logicalAnd).orElse(true)).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> addAll(Collection<? extends E> c) {
+ final MeteringAgent.Context timer = monitor.startTimer(ADD_ALL);
+ return Tools.allOf(c.stream().map(this::add).collect(Collectors.toList())).thenApply(v ->
+ v.stream().reduce(Boolean::logicalOr).orElse(false)).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> retainAll(Collection<? extends E> c) {
+ final MeteringAgent.Context timer = monitor.startTimer(RETAIN_ALL);
+ return backingMap.keySet().thenApply(set -> Sets.difference(set, Sets.newHashSet(c)))
+ .thenCompose(this::removeAll)
+ .whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> removeAll(Collection<? extends E> c) {
+ final MeteringAgent.Context timer = monitor.startTimer(REMOVE_ALL);
+ return Tools.allOf(c.stream().map(this::remove).collect(Collectors.toList())).thenApply(v ->
+ v.stream().reduce(Boolean::logicalOr).orElse(false)).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Void> clear() {
+ final MeteringAgent.Context timer = monitor.startTimer(CLEAR);
+ return backingMap.clear().whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<? extends Set<E>> getAsImmutableSet() {
+ final MeteringAgent.Context timer = monitor.startTimer(GET_AS_IMMUTABLE_SET);
+ return backingMap.keySet().thenApply(s -> ImmutableSet.copyOf(s)).whenComplete((r, e) -> timer.stop(null));
+ }
+
+ @Override
+ public CompletableFuture<Void> addListener(SetEventListener<E> listener) {
+ MapEventListener<E, Boolean> mapEventListener = mapEvent -> {
+ if (mapEvent.type() == MapEvent.Type.INSERT) {
+ listener.event(new SetEvent<>(name, SetEvent.Type.ADD, mapEvent.key()));
+ } else if (mapEvent.type() == MapEvent.Type.REMOVE) {
+ listener.event(new SetEvent<>(name, SetEvent.Type.REMOVE, mapEvent.key()));
+ }
+ };
+ if (listenerMapping.putIfAbsent(listener, mapEventListener) == null) {
+ return backingMap.addListener(mapEventListener);
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> removeListener(SetEventListener<E> listener) {
+ MapEventListener<E, Boolean> mapEventListener = listenerMapping.remove(listener);
+ if (mapEventListener != null) {
+ return backingMap.removeListener(mapEventListener);
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounter.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounter.java
new file mode 100644
index 0000000..74b6942
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounter.java
@@ -0,0 +1,92 @@
+/*
+ * 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.AtomicCounter;
+import org.onosproject.store.service.StorageException;
+import org.onosproject.store.service.Synchronous;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Default implementation for a distributed AtomicCounter backed by
+ * partitioned Raft DB.
+ * <p>
+ * The initial value will be zero.
+ */
+public class DefaultAtomicCounter extends Synchronous<AsyncAtomicCounter> implements AtomicCounter {
+
+ private static final int OPERATION_TIMEOUT_MILLIS = 5000;
+
+ private final AsyncAtomicCounter asyncCounter;
+
+ public DefaultAtomicCounter(AsyncAtomicCounter asyncCounter) {
+ super(asyncCounter);
+ this.asyncCounter = asyncCounter;
+ }
+
+ @Override
+ public long incrementAndGet() {
+ return complete(asyncCounter.incrementAndGet());
+ }
+
+ @Override
+ public long getAndIncrement() {
+ return complete(asyncCounter.getAndIncrement());
+ }
+
+ @Override
+ public long getAndAdd(long delta) {
+ return complete(asyncCounter.getAndAdd(delta));
+ }
+
+ @Override
+ public long addAndGet(long delta) {
+ return complete(asyncCounter.getAndAdd(delta));
+ }
+
+ @Override
+ public void set(long value) {
+ complete(asyncCounter.set(value));
+ }
+
+ @Override
+ public boolean compareAndSet(long expectedValue, long updateValue) {
+ return complete(asyncCounter.compareAndSet(expectedValue, updateValue));
+ }
+
+ @Override
+ public long get() {
+ return complete(asyncCounter.get());
+ }
+
+ private static <T> T complete(CompletableFuture<T> future) {
+ try {
+ return future.get(OPERATION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new StorageException.Interrupted();
+ } catch (TimeoutException e) {
+ throw new StorageException.Timeout();
+ } catch (ExecutionException e) {
+ throw new StorageException(e.getCause());
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounterBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounterBuilder.java
new file mode 100644
index 0000000..a7a9a55
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicCounterBuilder.java
@@ -0,0 +1,75 @@
+/*
+ * 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.AtomicCounter;
+import org.onosproject.store.service.AtomicCounterBuilder;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+/**
+ * Default implementation of AtomicCounterBuilder.
+ */
+public class DefaultAtomicCounterBuilder implements AtomicCounterBuilder {
+
+ private String name;
+ private boolean partitionsEnabled = true;
+ private final Database partitionedDatabase;
+ private final Database inMemoryDatabase;
+ private boolean metering = true;
+
+ public DefaultAtomicCounterBuilder(Database inMemoryDatabase, Database partitionedDatabase) {
+ this.inMemoryDatabase = inMemoryDatabase;
+ this.partitionedDatabase = partitionedDatabase;
+ }
+
+ @Override
+ public AtomicCounterBuilder withName(String name) {
+ checkArgument(name != null && !name.isEmpty());
+ this.name = name;
+ return this;
+ }
+
+ @Override
+ public AtomicCounterBuilder withPartitionsDisabled() {
+ partitionsEnabled = false;
+ return this;
+ }
+
+ @Override
+ public AtomicCounter build() {
+ return new DefaultAtomicCounter(buildAsyncCounter());
+ }
+
+ @Override
+ public AsyncAtomicCounter buildAsyncCounter() {
+ validateInputs();
+ Database database = partitionsEnabled ? partitionedDatabase : inMemoryDatabase;
+ return new DefaultAsyncAtomicCounter(name, database, metering);
+ }
+
+ @Override
+ public AtomicCounterBuilder withMeteringDisabled() {
+ metering = false;
+ return this;
+ }
+
+ private void validateInputs() {
+ checkState(name != null, "name must be specified");
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicValue.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicValue.java
new file mode 100644
index 0000000..45808ba
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicValue.java
@@ -0,0 +1,77 @@
+/*
+ * 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.concurrent.TimeUnit;
+
+import org.onosproject.store.service.AsyncAtomicValue;
+import org.onosproject.store.service.AtomicValue;
+import org.onosproject.store.service.AtomicValueEventListener;
+import org.onosproject.store.service.StorageException;
+import org.onosproject.store.service.Synchronous;
+
+import com.google.common.util.concurrent.Futures;
+
+/**
+ * Default implementation of {@link AtomicValue}.
+ *
+ * @param <V> value type
+ */
+public class DefaultAtomicValue<V> extends Synchronous<AsyncAtomicValue<V>> implements AtomicValue<V> {
+
+ private static final int OPERATION_TIMEOUT_MILLIS = 5000;
+ private final AsyncAtomicValue<V> asyncValue;
+
+ public DefaultAtomicValue(AsyncAtomicValue<V> asyncValue) {
+ super(asyncValue);
+ this.asyncValue = asyncValue;
+ }
+
+ @Override
+ public boolean compareAndSet(V expect, V update) {
+ return complete(asyncValue.compareAndSet(expect, update));
+ }
+
+ @Override
+ public V get() {
+ return complete(asyncValue.get());
+ }
+
+ @Override
+ public V getAndSet(V value) {
+ return complete(asyncValue.getAndSet(value));
+ }
+
+ @Override
+ public void set(V value) {
+ complete(asyncValue.set(value));
+ }
+
+ @Override
+ public void addListener(AtomicValueEventListener<V> listener) {
+ complete(asyncValue.addListener(listener));
+ }
+
+ @Override
+ public void removeListener(AtomicValueEventListener<V> listener) {
+ complete(asyncValue.removeListener(listener));
+ }
+
+ private static <V> V complete(CompletableFuture<V> future) {
+ return Futures.getChecked(future, StorageException.class, OPERATION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicValueBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicValueBuilder.java
new file mode 100644
index 0000000..e2f4971
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultAtomicValueBuilder.java
@@ -0,0 +1,76 @@
+/*
+ * 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.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncAtomicValue;
+import org.onosproject.store.service.AtomicValue;
+import org.onosproject.store.service.AtomicValueBuilder;
+import org.onosproject.store.service.ConsistentMapBuilder;
+import org.onosproject.store.service.Serializer;
+
+/**
+ * Default implementation of AtomicValueBuilder.
+ *
+ * @param <V> value type
+ */
+public class DefaultAtomicValueBuilder<V> implements AtomicValueBuilder<V> {
+
+ private String name;
+ private ConsistentMapBuilder<String, V> mapBuilder;
+ private boolean metering = true;
+
+ public DefaultAtomicValueBuilder(DatabaseManager manager) {
+ mapBuilder = manager.<String, V>consistentMapBuilder()
+ .withName("onos-atomic-values")
+ .withMeteringDisabled()
+ .withSerializer(Serializer.using(KryoNamespaces.BASIC));
+ }
+
+ @Override
+ public AtomicValueBuilder<V> withName(String name) {
+ this.name = name;
+ return this;
+ }
+
+ @Override
+ public AtomicValueBuilder<V> withSerializer(Serializer serializer) {
+ mapBuilder.withSerializer(serializer);
+ return this;
+ }
+
+ @Override
+ public AtomicValueBuilder<V> withPartitionsDisabled() {
+ mapBuilder.withPartitionsDisabled();
+ return this;
+ }
+
+ @Override
+ public AtomicValueBuilder<V> withMeteringDisabled() {
+ metering = false;
+ return this;
+ }
+
+ @Override
+ public AsyncAtomicValue<V> buildAsyncValue() {
+ return new DefaultAsyncAtomicValue<>(mapBuilder.buildAsyncMap(), name, metering);
+ }
+
+ @Override
+ public AtomicValue<V> build() {
+ return new DefaultAtomicValue<>(buildAsyncValue());
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMap.java
new file mode 100644
index 0000000..80e3133
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMap.java
@@ -0,0 +1,208 @@
+/*
+ * 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.Collection;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.Set;
+
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.ConsistentMapException;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Synchronous;
+import org.onosproject.store.service.Versioned;
+
+/**
+ * ConsistentMap implementation that is backed by a Raft consensus
+ * based database.
+ *
+ * @param <K> type of key.
+ * @param <V> type of value.
+ */
+public class DefaultConsistentMap<K, V> extends Synchronous<AsyncConsistentMap<K, V>> implements ConsistentMap<K, V> {
+
+ private static final int OPERATION_TIMEOUT_MILLIS = 5000;
+
+ private final DefaultAsyncConsistentMap<K, V> asyncMap;
+ private Map<K, V> javaMap;
+
+ public DefaultConsistentMap(DefaultAsyncConsistentMap<K, V> asyncMap) {
+ super(asyncMap);
+ this.asyncMap = asyncMap;
+ }
+
+ @Override
+ public int size() {
+ return complete(asyncMap.size());
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return complete(asyncMap.isEmpty());
+ }
+
+ @Override
+ public boolean containsKey(K key) {
+ return complete(asyncMap.containsKey(key));
+ }
+
+ @Override
+ public boolean containsValue(V value) {
+ return complete(asyncMap.containsValue(value));
+ }
+
+ @Override
+ public Versioned<V> get(K key) {
+ return complete(asyncMap.get(key));
+ }
+
+ @Override
+ public Versioned<V> computeIfAbsent(K key,
+ Function<? super K, ? extends V> mappingFunction) {
+ return complete(asyncMap.computeIfAbsent(key, mappingFunction));
+ }
+
+ @Override
+ public Versioned<V> computeIfPresent(K key,
+ BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+ return complete(asyncMap.computeIfPresent(key, remappingFunction));
+ }
+
+ @Override
+ public Versioned<V> compute(K key,
+ BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+ return complete(asyncMap.compute(key, remappingFunction));
+ }
+
+ @Override
+ public Versioned<V> computeIf(K key,
+ Predicate<? super V> condition,
+ BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
+ return complete(asyncMap.computeIf(key, condition, remappingFunction));
+ }
+
+ @Override
+ public Versioned<V> put(K key, V value) {
+ return complete(asyncMap.put(key, value));
+ }
+
+ @Override
+ public Versioned<V> putAndGet(K key, V value) {
+ return complete(asyncMap.putAndGet(key, value));
+ }
+
+ @Override
+ public Versioned<V> remove(K key) {
+ return complete(asyncMap.remove(key));
+ }
+
+ @Override
+ public void clear() {
+ complete(asyncMap.clear());
+ }
+
+ @Override
+ public Set<K> keySet() {
+ return complete(asyncMap.keySet());
+ }
+
+ @Override
+ public Collection<Versioned<V>> values() {
+ return complete(asyncMap.values());
+ }
+
+ @Override
+ public Set<Entry<K, Versioned<V>>> entrySet() {
+ return complete(asyncMap.entrySet());
+ }
+
+ @Override
+ public Versioned<V> putIfAbsent(K key, V value) {
+ return complete(asyncMap.putIfAbsent(key, value));
+ }
+
+ @Override
+ public boolean remove(K key, V value) {
+ return complete(asyncMap.remove(key, value));
+ }
+
+ @Override
+ public boolean remove(K key, long version) {
+ return complete(asyncMap.remove(key, version));
+ }
+
+ @Override
+ public Versioned<V> replace(K key, V value) {
+ return complete(asyncMap.replace(key, value));
+ }
+
+ @Override
+ public boolean replace(K key, V oldValue, V newValue) {
+ return complete(asyncMap.replace(key, oldValue, newValue));
+ }
+
+ @Override
+ public boolean replace(K key, long oldVersion, V newValue) {
+ return complete(asyncMap.replace(key, oldVersion, newValue));
+ }
+
+ private static <T> T complete(CompletableFuture<T> future) {
+ try {
+ return future.get(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) {
+ if (e.getCause() instanceof ConsistentMapException) {
+ throw (ConsistentMapException) e.getCause();
+ } else {
+ throw new ConsistentMapException(e.getCause());
+ }
+ }
+ }
+
+ @Override
+ public void addListener(MapEventListener<K, V> listener) {
+ asyncMap.addListener(listener);
+ }
+
+ @Override
+ public void removeListener(MapEventListener<K, V> listener) {
+ asyncMap.addListener(listener);
+ }
+
+ @Override
+ public Map<K, V> asJavaMap() {
+ synchronized (this) {
+ if (javaMap == null) {
+ javaMap = new ConsistentMapBackedJavaMap<>(this);
+ }
+ }
+ return javaMap;
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMapBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMapBuilder.java
new file mode 100644
index 0000000..cfd9425
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMapBuilder.java
@@ -0,0 +1,141 @@
+/*
+ * 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.core.ApplicationId;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.ConsistentMapBuilder;
+import org.onosproject.store.service.Serializer;
+
+import static com.google.common.base.Preconditions.checkArgument;
+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> implements ConsistentMapBuilder<K, V> {
+
+ private Serializer serializer;
+ private String name;
+ private ApplicationId applicationId;
+ private boolean purgeOnUninstall = false;
+ private boolean partitionsEnabled = true;
+ private boolean readOnly = false;
+ private boolean metering = true;
+ private boolean relaxedReadConsistency = false;
+ private final DatabaseManager manager;
+
+ public DefaultConsistentMapBuilder(DatabaseManager manager) {
+ this.manager = manager;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withName(String name) {
+ checkArgument(name != null && !name.isEmpty());
+ this.name = name;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withApplicationId(ApplicationId id) {
+ checkArgument(id != null);
+ this.applicationId = id;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withPurgeOnUninstall() {
+ purgeOnUninstall = true;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withMeteringDisabled() {
+ metering = false;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withSerializer(Serializer serializer) {
+ checkArgument(serializer != null);
+ this.serializer = serializer;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withPartitionsDisabled() {
+ partitionsEnabled = false;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withUpdatesDisabled() {
+ readOnly = true;
+ return this;
+ }
+
+ @Override
+ public ConsistentMapBuilder<K, V> withRelaxedReadConsistency() {
+ relaxedReadConsistency = true;
+ return this;
+ }
+
+ 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 new DefaultConsistentMap<>(buildAndRegisterMap());
+ }
+
+ @Override
+ public AsyncConsistentMap<K, V> buildAsyncMap() {
+ return buildAndRegisterMap();
+ }
+
+ private DefaultAsyncConsistentMap<K, V> buildAndRegisterMap() {
+ validateInputs();
+ Database database = partitionsEnabled ? manager.partitionedDatabase : manager.inMemoryDatabase;
+ if (relaxedReadConsistency) {
+ return manager.registerMap(
+ new AsyncCachingConsistentMap<>(name,
+ applicationId,
+ database,
+ serializer,
+ readOnly,
+ purgeOnUninstall,
+ metering));
+ } else {
+ return manager.registerMap(
+ new DefaultAsyncConsistentMap<>(name,
+ applicationId,
+ database,
+ serializer,
+ readOnly,
+ purgeOnUninstall,
+ metering));
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabase.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabase.java
new file mode 100644
index 0000000..53edcbf
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabase.java
@@ -0,0 +1,254 @@
+/*
+ * 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.Transaction;
+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));
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java
new file mode 100644
index 0000000..45869ff
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java
@@ -0,0 +1,375 @@
+/*
+ * 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.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
+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 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(DatabaseUpdate 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(DatabaseUpdate update, long 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(DatabaseUpdate update, long 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(DatabaseUpdate update, long 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, long 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 long transactionId;
+ private final byte[] value;
+
+ public Update(long txId, byte[] value) {
+ this.transactionId = txId;
+ this.value = value;
+ }
+
+ public long transactionId() {
+ return this.transactionId;
+ }
+
+ public byte[] value() {
+ return this.value;
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueue.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueue.java
new file mode 100644
index 0000000..8ea1d72
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueue.java
@@ -0,0 +1,137 @@
+/*
+ * 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 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 type() {
+ 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);
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueueBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueueBuilder.java
new file mode 100644
index 0000000..ea3665e
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedQueueBuilder.java
@@ -0,0 +1,81 @@
+/*
+ * 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);
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedSet.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedSet.java
new file mode 100644
index 0000000..39c9f07
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedSet.java
@@ -0,0 +1,152 @@
+/*
+ * 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.AsyncDistributedSet;
+import org.onosproject.store.service.ConsistentMapException;
+import org.onosproject.store.service.DistributedSet;
+import org.onosproject.store.service.SetEventListener;
+import org.onosproject.store.service.Synchronous;
+
+import java.lang.reflect.Array;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Implementation of {@link DistributedSet} that merely delegates to a {@link AsyncDistributedSet}
+ * and waits for the operation to complete.
+
+ * @param <E> set element type
+ */
+public class DefaultDistributedSet<E> extends Synchronous<AsyncDistributedSet<E>> implements DistributedSet<E> {
+
+ private static final long OPERATION_TIMEOUT_MILLIS = 5000;
+
+ private final AsyncDistributedSet<E> asyncSet;
+
+ public DefaultDistributedSet(AsyncDistributedSet<E> asyncSet) {
+ super(asyncSet);
+ this.asyncSet = asyncSet;
+ }
+
+ private static <T> T complete(CompletableFuture<T> future) {
+ try {
+ return future.get(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) {
+ if (e.getCause() instanceof ConsistentMapException) {
+ throw (ConsistentMapException) e.getCause();
+ } else {
+ throw new ConsistentMapException(e.getCause());
+ }
+ }
+ }
+
+ @Override
+ public int size() {
+ return complete(asyncSet.size());
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return complete(asyncSet.isEmpty());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean contains(Object o) {
+ return complete(asyncSet.contains((E) o));
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ return complete(asyncSet.getAsImmutableSet()).iterator();
+ }
+
+ @Override
+ public Object[] toArray() {
+ return complete(asyncSet.getAsImmutableSet()).stream().toArray();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> T[] toArray(T[] a) {
+ // TODO: Optimize this to only allocate a new array if the set size
+ // is larger than the array.length. If the set size is smaller than
+ // the array.length then copy the data into the array and set the
+ // last element in the array to be null.
+ final T[] resizedArray =
+ (T[]) Array.newInstance(a.getClass().getComponentType(), complete(asyncSet.getAsImmutableSet()).size());
+ return complete(asyncSet.getAsImmutableSet()).toArray(resizedArray);
+ }
+
+ @Override
+ public boolean add(E e) {
+ return complete(asyncSet.add(e));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean remove(Object o) {
+ return complete(asyncSet.remove((E) o));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ return complete(asyncSet.containsAll((Collection<? extends E>) c));
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ return complete(asyncSet.addAll(c));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ return complete(asyncSet.retainAll((Collection<? extends E>) c));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ return complete(asyncSet.removeAll((Collection<? extends E>) c));
+ }
+
+ @Override
+ public void clear() {
+ complete(asyncSet.clear());
+ }
+
+ @Override
+ public void addListener(SetEventListener<E> listener) {
+ complete(asyncSet.addListener(listener));
+ }
+
+ @Override
+ public void removeListener(SetEventListener<E> listener) {
+ complete(asyncSet.removeListener(listener));
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedSetBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedSetBuilder.java
new file mode 100644
index 0000000..304044c
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDistributedSetBuilder.java
@@ -0,0 +1,99 @@
+/*
+ * 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.core.ApplicationId;
+import org.onosproject.store.service.AsyncDistributedSet;
+import org.onosproject.store.service.ConsistentMapBuilder;
+import org.onosproject.store.service.DistributedSet;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.DistributedSetBuilder;
+
+/**
+ * Default distributed set builder.
+ *
+ * @param <E> type for set elements
+ */
+public class DefaultDistributedSetBuilder<E> implements DistributedSetBuilder<E> {
+
+ private String name;
+ private ConsistentMapBuilder<E, Boolean> mapBuilder;
+ private boolean metering = true;
+
+ public DefaultDistributedSetBuilder(DatabaseManager manager) {
+ this.mapBuilder = manager.consistentMapBuilder();
+ mapBuilder.withMeteringDisabled();
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withName(String name) {
+ mapBuilder.withName(name);
+ this.name = name;
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withApplicationId(ApplicationId id) {
+ mapBuilder.withApplicationId(id);
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withPurgeOnUninstall() {
+ mapBuilder.withPurgeOnUninstall();
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withSerializer(Serializer serializer) {
+ mapBuilder.withSerializer(serializer);
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withUpdatesDisabled() {
+ mapBuilder.withUpdatesDisabled();
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withRelaxedReadConsistency() {
+ mapBuilder.withRelaxedReadConsistency();
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withPartitionsDisabled() {
+ mapBuilder.withPartitionsDisabled();
+ return this;
+ }
+
+ @Override
+ public DistributedSetBuilder<E> withMeteringDisabled() {
+ metering = false;
+ return this;
+ }
+
+ @Override
+ public DistributedSet<E> build() {
+ return new DefaultDistributedSet<E>(buildAsyncSet());
+ }
+
+ @Override
+ public AsyncDistributedSet<E> buildAsyncSet() {
+ return new DefaultAsyncDistributedSet<E>(mapBuilder.buildAsyncMap(), name, metering);
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransaction.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransaction.java
new file mode 100644
index 0000000..406e901
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransaction.java
@@ -0,0 +1,70 @@
+/*
+ * 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 org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * A Default transaction implementation.
+ */
+public class DefaultTransaction implements Transaction {
+
+ private final long transactionId;
+ private final List<DatabaseUpdate> updates;
+ private final State state;
+ private final long lastUpdated;
+
+ public DefaultTransaction(long transactionId, List<DatabaseUpdate> updates) {
+ this(transactionId, updates, State.PREPARING, System.currentTimeMillis());
+ }
+
+ private DefaultTransaction(long transactionId, List<DatabaseUpdate> updates, State state, long lastUpdated) {
+ this.transactionId = transactionId;
+ this.updates = ImmutableList.copyOf(updates);
+ this.state = state;
+ this.lastUpdated = lastUpdated;
+ }
+
+ @Override
+ public long id() {
+ return transactionId;
+ }
+
+ @Override
+ public List<DatabaseUpdate> updates() {
+ return updates;
+ }
+
+ @Override
+ public State state() {
+ return state;
+ }
+
+ @Override
+ public Transaction transition(State newState) {
+ return new DefaultTransaction(transactionId, updates, newState, System.currentTimeMillis());
+ }
+
+ @Override
+ public long lastUpdated() {
+ return lastUpdated;
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java
new file mode 100644
index 0000000..c8abc04
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContext.java
@@ -0,0 +1,117 @@
+/*
+ * 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.function.Supplier;
+
+import static com.google.common.base.Preconditions.*;
+
+import org.onosproject.store.service.ConsistentMapBuilder;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.Transaction;
+import org.onosproject.store.service.TransactionContext;
+import org.onosproject.store.service.TransactionalMap;
+
+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 Database database;
+ private final long transactionId;
+ private final Supplier<ConsistentMapBuilder> mapBuilderSupplier;
+
+ public DefaultTransactionContext(long transactionId,
+ Database database,
+ Supplier<ConsistentMapBuilder> mapBuilderSupplier) {
+ this.transactionId = transactionId;
+ this.database = checkNotNull(database);
+ this.mapBuilderSupplier = checkNotNull(mapBuilderSupplier);
+ }
+
+ @Override
+ public long 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 -> new DefaultTransactionalMap<>(
+ name,
+ mapBuilderSupplier.get().withName(name).withSerializer(serializer).build(),
+ this,
+ serializer));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean commit() {
+ // TODO: rework commit implementation to be more intuitive
+ checkState(isOpen, TX_NOT_OPEN_ERROR);
+ CommitResponse response = null;
+ try {
+ List<DatabaseUpdate> updates = Lists.newLinkedList();
+ txMaps.values().forEach(m -> updates.addAll(m.prepareDatabaseUpdates()));
+ Transaction transaction = new DefaultTransaction(transactionId, updates);
+ response = Futures.getUnchecked(database.prepareAndCommit(transaction));
+ return response.success();
+ } catch (Exception e) {
+ abort();
+ return false;
+ } finally {
+ isOpen = false;
+ }
+ }
+
+ @Override
+ public void abort() {
+ if (isOpen) {
+ try {
+ txMaps.values().forEach(m -> m.rollback());
+ } finally {
+ isOpen = false;
+ }
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java
new file mode 100644
index 0000000..36b2831
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionContextBuilder.java
@@ -0,0 +1,50 @@
+/*
+ * 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.TransactionContext;
+import org.onosproject.store.service.TransactionContextBuilder;
+
+/**
+ * The default implementation of a transaction context builder. This builder
+ * generates a {@link DefaultTransactionContext}.
+ */
+public class DefaultTransactionContextBuilder implements TransactionContextBuilder {
+
+ private boolean partitionsEnabled = true;
+ private final DatabaseManager manager;
+ private final long transactionId;
+
+ public DefaultTransactionContextBuilder(DatabaseManager manager, long transactionId) {
+ this.manager = manager;
+ this.transactionId = transactionId;
+ }
+
+ @Override
+ public TransactionContextBuilder withPartitionsDisabled() {
+ partitionsEnabled = false;
+ return this;
+ }
+
+ @Override
+ public TransactionContext build() {
+ return new DefaultTransactionContext(
+ transactionId,
+ partitionsEnabled ? manager.partitionedDatabase : manager.inMemoryDatabase,
+ () -> partitionsEnabled ? manager.consistentMapBuilder()
+ : manager.consistentMapBuilder().withPartitionsDisabled());
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMap.java
new file mode 100644
index 0000000..3594adb
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalMap.java
@@ -0,0 +1,204 @@
+/*
+ * 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.Set;
+
+import org.onlab.util.HexString;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.TransactionContext;
+import org.onosproject.store.service.TransactionalMap;
+import org.onosproject.store.service.Versioned;
+
+import static com.google.common.base.Preconditions.*;
+
+import com.google.common.base.Objects;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+/**
+ * Default Transactional Map implementation that provides a repeatable reads
+ * transaction isolation level.
+ *
+ * @param <K> key type
+ * @param <V> value type.
+ */
+public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> {
+
+ private final TransactionContext txContext;
+ private static final String TX_CLOSED_ERROR = "Transaction is closed";
+ private final ConsistentMap<K, V> backingMap;
+ private final String name;
+ private final Serializer serializer;
+ private final Map<K, Versioned<V>> readCache = Maps.newConcurrentMap();
+ private final Map<K, V> writeCache = Maps.newConcurrentMap();
+ private final Set<K> deleteSet = Sets.newConcurrentHashSet();
+
+ private static final String ERROR_NULL_VALUE = "Null values are not allowed";
+ private static final String ERROR_NULL_KEY = "Null key is not allowed";
+
+ private final LoadingCache<K, String> keyCache = CacheBuilder.newBuilder()
+ .softValues()
+ .build(new CacheLoader<K, String>() {
+
+ @Override
+ public String load(K key) {
+ return HexString.toHexString(serializer.encode(key));
+ }
+ });
+
+ protected K dK(String key) {
+ return serializer.decode(HexString.fromHexString(key));
+ }
+
+ public DefaultTransactionalMap(
+ String name,
+ ConsistentMap<K, V> backingMap,
+ TransactionContext txContext,
+ Serializer serializer) {
+ this.name = name;
+ this.backingMap = backingMap;
+ this.txContext = txContext;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public V get(K key) {
+ checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+ checkNotNull(key, ERROR_NULL_KEY);
+ if (deleteSet.contains(key)) {
+ return null;
+ }
+ V latest = writeCache.get(key);
+ if (latest != null) {
+ return latest;
+ } else {
+ Versioned<V> v = readCache.computeIfAbsent(key, k -> backingMap.get(k));
+ return v != null ? v.value() : null;
+ }
+ }
+
+ @Override
+ public V put(K key, V value) {
+ checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+ checkNotNull(value, ERROR_NULL_VALUE);
+
+ V latest = get(key);
+ writeCache.put(key, value);
+ deleteSet.remove(key);
+ return latest;
+ }
+
+ @Override
+ public V remove(K key) {
+ checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+ V latest = get(key);
+ if (latest != null) {
+ writeCache.remove(key);
+ deleteSet.add(key);
+ }
+ return latest;
+ }
+
+ @Override
+ public boolean remove(K key, V value) {
+ checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+ checkNotNull(value, ERROR_NULL_VALUE);
+ V latest = get(key);
+ if (Objects.equal(value, latest)) {
+ remove(key);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean replace(K key, V oldValue, V newValue) {
+ checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+ checkNotNull(oldValue, ERROR_NULL_VALUE);
+ checkNotNull(newValue, ERROR_NULL_VALUE);
+ V latest = get(key);
+ if (Objects.equal(oldValue, latest)) {
+ put(key, newValue);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public V putIfAbsent(K key, V value) {
+ checkState(txContext.isOpen(), TX_CLOSED_ERROR);
+ checkNotNull(value, ERROR_NULL_VALUE);
+ V latest = get(key);
+ if (latest == null) {
+ put(key, value);
+ }
+ return latest;
+ }
+
+ protected List<DatabaseUpdate> prepareDatabaseUpdates() {
+ List<DatabaseUpdate> updates = Lists.newLinkedList();
+ deleteSet.forEach(key -> {
+ Versioned<V> original = readCache.get(key);
+ if (original != null) {
+ updates.add(DatabaseUpdate.newBuilder()
+ .withMapName(name)
+ .withType(DatabaseUpdate.Type.REMOVE_IF_VERSION_MATCH)
+ .withKey(keyCache.getUnchecked(key))
+ .withCurrentVersion(original.version())
+ .build());
+ }
+ });
+ writeCache.forEach((key, value) -> {
+ Versioned<V> original = readCache.get(key);
+ if (original == null) {
+ updates.add(DatabaseUpdate.newBuilder()
+ .withMapName(name)
+ .withType(DatabaseUpdate.Type.PUT_IF_ABSENT)
+ .withKey(keyCache.getUnchecked(key))
+ .withValue(serializer.encode(value))
+ .build());
+ } else {
+ updates.add(DatabaseUpdate.newBuilder()
+ .withMapName(name)
+ .withType(DatabaseUpdate.Type.PUT_IF_VERSION_MATCH)
+ .withKey(keyCache.getUnchecked(key))
+ .withCurrentVersion(original.version())
+ .withValue(serializer.encode(value))
+ .build());
+ }
+ });
+ return updates;
+ }
+
+ /**
+ * Discards all changes made to this transactional map.
+ */
+ protected void rollback() {
+ readCache.clear();
+ writeCache.clear();
+ deleteSet.clear();
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalSet.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalSet.java
new file mode 100644
index 0000000..f03e59e
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultTransactionalSet.java
@@ -0,0 +1,58 @@
+/*
+ * 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.Serializer;
+import org.onosproject.store.service.TransactionContext;
+import org.onosproject.store.service.TransactionalMap;
+import org.onosproject.store.service.TransactionalSet;
+
+/**
+ * Default TransactionalSet implementation that provides a repeatable reads
+ * transaction isolation level.
+ *
+ * @param <E> element type.
+ */
+public class DefaultTransactionalSet<E> implements TransactionalSet<E> {
+
+ private TransactionalMap<E, Boolean> map;
+
+ // dummy value to associate with an Object in the backing map
+ private static final Boolean PRESENT = new Boolean(true);
+
+ public DefaultTransactionalSet(
+ String name,
+ TransactionContext txContext,
+ Serializer serializer) {
+ map = txContext.getTransactionalMap(name, serializer);
+ }
+
+ @Override
+ public boolean add(E e) {
+ return map.put(e, PRESENT) == null;
+ }
+
+ @Override
+ public boolean remove(E e) {
+ return map.remove(e) != null;
+ }
+
+ @Override
+ public boolean contains(E e) {
+ return map.get(e) != null;
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DistributedLeadershipManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DistributedLeadershipManager.java
new file mode 100644
index 0000000..db55a85
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DistributedLeadershipManager.java
@@ -0,0 +1,614 @@
+/*
+ * 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.ImmutableMap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import org.apache.commons.lang.math.RandomUtils;
+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.Service;
+import org.onosproject.cluster.ClusterEvent;
+import org.onosproject.cluster.ClusterEvent.Type;
+import org.onosproject.cluster.ClusterEventListener;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.Leadership;
+import org.onosproject.cluster.LeadershipEvent;
+import org.onosproject.cluster.LeadershipEventListener;
+import org.onosproject.cluster.LeadershipService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.event.ListenerRegistry;
+import org.onosproject.event.EventDeliveryService;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.ConsistentMapException;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Set;
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.slf4j.LoggerFactory.getLogger;
+import static org.onosproject.cluster.ControllerNode.State.ACTIVE;
+import static org.onosproject.cluster.ControllerNode.State.INACTIVE;
+
+/**
+ * Distributed Lock Manager implemented on top of ConsistentMap.
+ * <p>
+ * This implementation makes use of ClusterService's failure
+ * detection capabilities to detect and purge stale locks.
+ * TODO: Ensure lock safety and liveness.
+ */
+@Component(immediate = true, enabled = true)
+@Service
+public class DistributedLeadershipManager implements LeadershipService {
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StorageService storageService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterCommunicationService clusterCommunicator;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected EventDeliveryService eventDispatcher;
+
+ private final Logger log = getLogger(getClass());
+
+ private ScheduledExecutorService electionRunner;
+ private ScheduledExecutorService lockExecutor;
+ private ScheduledExecutorService staleLeadershipPurgeExecutor;
+ private ScheduledExecutorService leadershipRefresher;
+
+ // leader for each topic
+ private ConsistentMap<String, NodeId> leaderMap;
+ // list of candidates (includes chosen leader) for each topic
+ private ConsistentMap<String, List<NodeId>> candidateMap;
+
+ private ListenerRegistry<LeadershipEvent, LeadershipEventListener> listenerRegistry;
+
+ // cached copy of leaderMap
+ // Note: Map value, Leadership, does not contain proper candidates info
+ private final Map<String, Leadership> leaderBoard = Maps.newConcurrentMap();
+ // cached copy of candidateMap
+ // Note: Map value, Leadership, does not contain proper leader info
+ private final Map<String, Leadership> candidateBoard = Maps.newConcurrentMap();
+
+ private final ClusterEventListener clusterEventListener = new InternalClusterEventListener();
+
+ private NodeId localNodeId;
+ private Set<String> activeTopics = Sets.newConcurrentHashSet();
+ private Map<String, CompletableFuture<Leadership>> pendingFutures = Maps.newConcurrentMap();
+
+ // The actual delay is randomly chosen from the interval [0, WAIT_BEFORE_RETRY_MILLIS)
+ private static final int WAIT_BEFORE_RETRY_MILLIS = 150;
+ private static final int DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC = 2;
+ private static final int LEADERSHIP_REFRESH_INTERVAL_SEC = 2;
+ private static final int DELAY_BETWEEN_STALE_LEADERSHIP_PURGE_ATTEMPTS_SEC = 2;
+
+ private final AtomicBoolean staleLeadershipPurgeScheduled = new AtomicBoolean(false);
+
+ private static final Serializer SERIALIZER = Serializer.using(KryoNamespaces.API);
+
+ @Activate
+ public void activate() {
+ leaderMap = storageService.<String, NodeId>consistentMapBuilder()
+ .withName("onos-topic-leaders")
+ .withSerializer(SERIALIZER)
+ .withPartitionsDisabled().build();
+ candidateMap = storageService.<String, List<NodeId>>consistentMapBuilder()
+ .withName("onos-topic-candidates")
+ .withSerializer(SERIALIZER)
+ .withPartitionsDisabled().build();
+
+ leaderMap.addListener(event -> {
+ log.debug("Received {}", event);
+ LeadershipEvent.Type leadershipEventType = null;
+ if (event.type() == MapEvent.Type.INSERT || event.type() == MapEvent.Type.UPDATE) {
+ leadershipEventType = LeadershipEvent.Type.LEADER_ELECTED;
+ } else if (event.type() == MapEvent.Type.REMOVE) {
+ leadershipEventType = LeadershipEvent.Type.LEADER_BOOTED;
+ }
+ onLeadershipEvent(new LeadershipEvent(
+ leadershipEventType,
+ new Leadership(event.key(),
+ event.value().value(),
+ event.value().version(),
+ event.value().creationTime())));
+ });
+
+ candidateMap.addListener(event -> {
+ log.debug("Received {}", event);
+ if (event.type() != MapEvent.Type.INSERT && event.type() != MapEvent.Type.UPDATE) {
+ log.error("Entries must not be removed from candidate map");
+ return;
+ }
+ onLeadershipEvent(new LeadershipEvent(
+ LeadershipEvent.Type.CANDIDATES_CHANGED,
+ new Leadership(event.key(),
+ event.value().value(),
+ event.value().version(),
+ event.value().creationTime())));
+ });
+
+ localNodeId = clusterService.getLocalNode().id();
+
+ electionRunner = Executors.newSingleThreadScheduledExecutor(
+ groupedThreads("onos/store/leadership", "election-runner"));
+ lockExecutor = Executors.newScheduledThreadPool(
+ 4, groupedThreads("onos/store/leadership", "election-thread-%d"));
+ staleLeadershipPurgeExecutor = Executors.newSingleThreadScheduledExecutor(
+ groupedThreads("onos/store/leadership", "stale-leadership-evictor"));
+ leadershipRefresher = Executors.newSingleThreadScheduledExecutor(
+ groupedThreads("onos/store/leadership", "refresh-thread"));
+
+ clusterService.addListener(clusterEventListener);
+
+ electionRunner.scheduleWithFixedDelay(
+ this::electLeaders, 0, DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC, TimeUnit.SECONDS);
+
+ leadershipRefresher.scheduleWithFixedDelay(
+ this::refreshLeaderBoard, 0, LEADERSHIP_REFRESH_INTERVAL_SEC, TimeUnit.SECONDS);
+
+ listenerRegistry = new ListenerRegistry<>();
+ eventDispatcher.addSink(LeadershipEvent.class, listenerRegistry);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ if (clusterService.getNodes().size() > 1) {
+ // FIXME: Determine why this takes ~50 seconds to shutdown on a single node!
+ leaderBoard.forEach((topic, leadership) -> {
+ if (localNodeId.equals(leadership.leader())) {
+ withdraw(topic);
+ }
+ });
+ }
+
+ clusterService.removeListener(clusterEventListener);
+ eventDispatcher.removeSink(LeadershipEvent.class);
+
+ electionRunner.shutdown();
+ lockExecutor.shutdown();
+ staleLeadershipPurgeExecutor.shutdown();
+ leadershipRefresher.shutdown();
+
+ log.info("Stopped");
+ }
+
+ @Override
+ public Map<String, Leadership> getLeaderBoard() {
+ return ImmutableMap.copyOf(leaderBoard);
+ }
+
+ @Override
+ public Map<String, List<NodeId>> getCandidates() {
+ return Maps.toMap(candidateBoard.keySet(), this::getCandidates);
+ }
+
+ @Override
+ public List<NodeId> getCandidates(String path) {
+ Leadership current = candidateBoard.get(path);
+ return current == null ? ImmutableList.of() : ImmutableList.copyOf(current.candidates());
+ }
+
+ @Override
+ public NodeId getLeader(String path) {
+ Leadership leadership = leaderBoard.get(path);
+ return leadership != null ? leadership.leader() : null;
+ }
+
+ @Override
+ public Leadership getLeadership(String path) {
+ checkArgument(path != null);
+ return leaderBoard.get(path);
+ }
+
+ @Override
+ public Set<String> ownedTopics(NodeId nodeId) {
+ checkArgument(nodeId != null);
+ return leaderBoard.entrySet()
+ .stream()
+ .filter(entry -> nodeId.equals(entry.getValue().leader()))
+ .map(Entry::getKey)
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public CompletableFuture<Leadership> runForLeadership(String path) {
+ log.debug("Running for leadership for topic: {}", path);
+ CompletableFuture<Leadership> resultFuture = new CompletableFuture<>();
+ doRunForLeadership(path, resultFuture);
+ return resultFuture;
+ }
+
+ private void doRunForLeadership(String path, CompletableFuture<Leadership> future) {
+ try {
+ Versioned<List<NodeId>> candidates = candidateMap.computeIf(path,
+ currentList -> currentList == null || !currentList.contains(localNodeId),
+ (topic, currentList) -> {
+ if (currentList == null) {
+ return ImmutableList.of(localNodeId);
+ } else {
+ List<NodeId> newList = Lists.newLinkedList();
+ newList.addAll(currentList);
+ newList.add(localNodeId);
+ return newList;
+ }
+ });
+ log.debug("In the leadership race for topic {} with candidates {}", path, candidates);
+ activeTopics.add(path);
+ Leadership leadership = electLeader(path, candidates.value());
+ if (leadership == null) {
+ pendingFutures.put(path, future);
+ } else {
+ future.complete(leadership);
+ }
+ } catch (ConsistentMapException e) {
+ log.debug("Failed to enter topic leader race for {}. Retrying.", path, e);
+ rerunForLeadership(path, future);
+ }
+ }
+
+ @Override
+ public CompletableFuture<Void> withdraw(String path) {
+ activeTopics.remove(path);
+ CompletableFuture<Void> resultFuture = new CompletableFuture<>();
+ doWithdraw(path, resultFuture);
+ return resultFuture;
+ }
+
+
+ private void doWithdraw(String path, CompletableFuture<Void> future) {
+ if (activeTopics.contains(path)) {
+ future.completeExceptionally(new CancellationException(String.format("%s is now a active topic", path)));
+ }
+ try {
+ leaderMap.computeIf(path,
+ localNodeId::equals,
+ (topic, leader) -> null);
+ candidateMap.computeIf(path,
+ candidates -> candidates != null && candidates.contains(localNodeId),
+ (topic, candidates) -> candidates.stream()
+ .filter(nodeId -> !localNodeId.equals(nodeId))
+ .collect(Collectors.toList()));
+ future.complete(null);
+ } catch (Exception e) {
+ log.debug("Failed to verify (and clear) any lock this node might be holding for {}", path, e);
+ retryWithdraw(path, future);
+ }
+ }
+
+ @Override
+ public boolean stepdown(String path) {
+ if (!activeTopics.contains(path) || !Objects.equals(localNodeId, getLeader(path))) {
+ return false;
+ }
+
+ try {
+ return leaderMap.computeIf(path,
+ localNodeId::equals,
+ (topic, leader) -> null) == null;
+ } catch (Exception e) {
+ log.warn("Error executing stepdown for {}", path, e);
+ }
+ return false;
+ }
+
+ @Override
+ public void addListener(LeadershipEventListener listener) {
+ listenerRegistry.addListener(listener);
+ }
+
+ @Override
+ public void removeListener(LeadershipEventListener listener) {
+ listenerRegistry.removeListener(listener);
+ }
+
+ @Override
+ public boolean makeTopCandidate(String path, NodeId nodeId) {
+ Versioned<List<NodeId>> candidateList = candidateMap.computeIf(path,
+ candidates -> candidates != null &&
+ candidates.contains(nodeId) &&
+ !nodeId.equals(Iterables.getFirst(candidates, null)),
+ (topic, candidates) -> {
+ List<NodeId> updatedCandidates = new ArrayList<>(candidates.size());
+ updatedCandidates.add(nodeId);
+ candidates.stream().filter(id -> !nodeId.equals(id)).forEach(updatedCandidates::add);
+ return updatedCandidates;
+ });
+ List<NodeId> candidates = candidateList != null ? candidateList.value() : Collections.emptyList();
+ return candidates.size() > 0 && nodeId.equals(candidates.get(0));
+ }
+
+ private Leadership electLeader(String path, List<NodeId> candidates) {
+ Leadership currentLeadership = getLeadership(path);
+ if (currentLeadership != null) {
+ return currentLeadership;
+ } else {
+ NodeId topCandidate = candidates
+ .stream()
+ .filter(n -> clusterService.getState(n) == ACTIVE)
+ .findFirst()
+ .orElse(null);
+ try {
+ Versioned<NodeId> leader = localNodeId.equals(topCandidate)
+ ? leaderMap.computeIfAbsent(path, p -> localNodeId) : leaderMap.get(path);
+ if (leader != null) {
+ Leadership newLeadership = new Leadership(path,
+ leader.value(),
+ leader.version(),
+ leader.creationTime());
+ // Since reads only go through the local copy of leader board, we ought to update it
+ // first before returning from this method.
+ // This is to ensure a subsequent read will not read a stale value.
+ onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, newLeadership));
+ return newLeadership;
+ }
+ } catch (Exception e) {
+ log.debug("Failed to elect leader for {}", path, e);
+ }
+ }
+ return null;
+ }
+
+ private void electLeaders() {
+ try {
+ candidateMap.entrySet().forEach(entry -> {
+ String path = entry.getKey();
+ Versioned<List<NodeId>> candidates = entry.getValue();
+ // for active topics, check if this node can become a leader (if it isn't already)
+ if (activeTopics.contains(path)) {
+ lockExecutor.submit(() -> {
+ Leadership leadership = electLeader(path, candidates.value());
+ if (leadership != null) {
+ CompletableFuture<Leadership> future = pendingFutures.remove(path);
+ if (future != null) {
+ future.complete(leadership);
+ }
+ }
+ });
+ }
+ // Raise a CANDIDATES_CHANGED event to force refresh local candidate board
+ // and also to update local listeners.
+ // Don't worry about duplicate events as they will be suppressed.
+ onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.CANDIDATES_CHANGED,
+ new Leadership(path,
+ candidates.value(),
+ candidates.version(),
+ candidates.creationTime())));
+ });
+ } catch (Exception e) {
+ log.debug("Failure electing leaders", e);
+ }
+ }
+
+ private void onLeadershipEvent(LeadershipEvent leadershipEvent) {
+ log.trace("Leadership Event: time = {} type = {} event = {}",
+ leadershipEvent.time(), leadershipEvent.type(),
+ leadershipEvent);
+
+ Leadership leadershipUpdate = leadershipEvent.subject();
+ LeadershipEvent.Type eventType = leadershipEvent.type();
+ String topic = leadershipUpdate.topic();
+
+ AtomicBoolean updateAccepted = new AtomicBoolean(false);
+ if (eventType.equals(LeadershipEvent.Type.LEADER_ELECTED)) {
+ leaderBoard.compute(topic, (k, currentLeadership) -> {
+ if (currentLeadership == null || currentLeadership.epoch() < leadershipUpdate.epoch()) {
+ updateAccepted.set(true);
+ return leadershipUpdate;
+ }
+ return currentLeadership;
+ });
+ } else if (eventType.equals(LeadershipEvent.Type.LEADER_BOOTED)) {
+ leaderBoard.compute(topic, (k, currentLeadership) -> {
+ if (currentLeadership == null || currentLeadership.epoch() <= leadershipUpdate.epoch()) {
+ updateAccepted.set(true);
+ // FIXME: Removing entries from leaderboard is not safe and should be visited.
+ return null;
+ }
+ return currentLeadership;
+ });
+ } else if (eventType.equals(LeadershipEvent.Type.CANDIDATES_CHANGED)) {
+ candidateBoard.compute(topic, (k, currentInfo) -> {
+ if (currentInfo == null || currentInfo.epoch() < leadershipUpdate.epoch()) {
+ updateAccepted.set(true);
+ return leadershipUpdate;
+ }
+ return currentInfo;
+ });
+ } else {
+ throw new IllegalStateException("Unknown event type.");
+ }
+
+ if (updateAccepted.get()) {
+ eventDispatcher.post(leadershipEvent);
+ }
+ }
+
+ private void rerunForLeadership(String path, CompletableFuture<Leadership> future) {
+ lockExecutor.schedule(
+ () -> doRunForLeadership(path, future),
+ RandomUtils.nextInt(WAIT_BEFORE_RETRY_MILLIS),
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void retryWithdraw(String path, CompletableFuture<Void> future) {
+ lockExecutor.schedule(
+ () -> doWithdraw(path, future),
+ RandomUtils.nextInt(WAIT_BEFORE_RETRY_MILLIS),
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void scheduleStaleLeadershipPurge(int afterDelaySec) {
+ if (staleLeadershipPurgeScheduled.compareAndSet(false, true)) {
+ staleLeadershipPurgeExecutor.schedule(
+ this::purgeStaleLeadership,
+ afterDelaySec,
+ TimeUnit.SECONDS);
+ }
+ }
+
+ /**
+ * Purges locks held by inactive nodes and evicts inactive nodes from candidacy.
+ */
+ private void purgeStaleLeadership() {
+ AtomicBoolean rerunPurge = new AtomicBoolean(false);
+ try {
+ staleLeadershipPurgeScheduled.set(false);
+ leaderMap.entrySet()
+ .stream()
+ .filter(e -> clusterService.getState(e.getValue().value()) == INACTIVE)
+ .forEach(entry -> {
+ String path = entry.getKey();
+ NodeId nodeId = entry.getValue().value();
+ try {
+ leaderMap.computeIf(path, nodeId::equals, (topic, leader) -> null);
+ } catch (Exception e) {
+ log.debug("Failed to purge stale lock held by {} for {}", nodeId, path, e);
+ rerunPurge.set(true);
+ }
+ });
+
+ candidateMap.entrySet()
+ .forEach(entry -> {
+ String path = entry.getKey();
+ Versioned<List<NodeId>> candidates = entry.getValue();
+ List<NodeId> candidatesList = candidates != null
+ ? candidates.value() : Collections.emptyList();
+ List<NodeId> activeCandidatesList =
+ candidatesList.stream()
+ .filter(n -> clusterService.getState(n) == ACTIVE)
+ .filter(n -> !localNodeId.equals(n) || activeTopics.contains(path))
+ .collect(Collectors.toList());
+ if (activeCandidatesList.size() < candidatesList.size()) {
+ Set<NodeId> removedCandidates =
+ Sets.difference(Sets.newHashSet(candidatesList),
+ Sets.newHashSet(activeCandidatesList));
+ try {
+ candidateMap.computeIf(path,
+ c -> c.stream()
+ .filter(n -> clusterService.getState(n) == INACTIVE)
+ .count() > 0,
+ (topic, c) -> c.stream()
+ .filter(n -> clusterService.getState(n) == ACTIVE)
+ .filter(n -> !localNodeId.equals(n) ||
+ activeTopics.contains(path))
+ .collect(Collectors.toList()));
+ } catch (Exception e) {
+ log.debug("Failed to evict inactive candidates {} from "
+ + "candidate list for {}", removedCandidates, path, e);
+ rerunPurge.set(true);
+ }
+ }
+ });
+ } catch (Exception e) {
+ log.debug("Failure purging state leadership.", e);
+ rerunPurge.set(true);
+ }
+
+ if (rerunPurge.get()) {
+ log.debug("Rescheduling stale leadership purge due to errors encountered in previous run");
+ scheduleStaleLeadershipPurge(DELAY_BETWEEN_STALE_LEADERSHIP_PURGE_ATTEMPTS_SEC);
+ }
+ }
+
+ private void refreshLeaderBoard() {
+ try {
+ Map<String, Leadership> newLeaderBoard = Maps.newHashMap();
+ leaderMap.entrySet().forEach(entry -> {
+ String path = entry.getKey();
+ Versioned<NodeId> leader = entry.getValue();
+ Leadership leadership = new Leadership(path,
+ leader.value(),
+ leader.version(),
+ leader.creationTime());
+ newLeaderBoard.put(path, leadership);
+ });
+
+ // first take snapshot of current leader board.
+ Map<String, Leadership> currentLeaderBoard = ImmutableMap.copyOf(leaderBoard);
+
+ MapDifference<String, Leadership> diff = Maps.difference(currentLeaderBoard, newLeaderBoard);
+
+ // evict stale leaders
+ diff.entriesOnlyOnLeft().forEach((path, leadership) -> {
+ log.debug("Evicting {} from leaderboard. It is no longer active leader.", leadership);
+ onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.LEADER_BOOTED, leadership));
+ });
+
+ // add missing leaders
+ diff.entriesOnlyOnRight().forEach((path, leadership) -> {
+ log.debug("Adding {} to leaderboard. It is now the active leader.", leadership);
+ onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, leadership));
+ });
+
+ // add updated leaders
+ diff.entriesDiffering().forEach((path, difference) -> {
+ Leadership current = difference.leftValue();
+ Leadership updated = difference.rightValue();
+ if (current.epoch() < updated.epoch()) {
+ log.debug("Updated {} in leaderboard.", updated);
+ onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, updated));
+ }
+ });
+ } catch (Exception e) {
+ log.debug("Failed to refresh leader board", e);
+ }
+ }
+
+ private class InternalClusterEventListener implements ClusterEventListener {
+
+ @Override
+ public void event(ClusterEvent event) {
+ if (event.type() == Type.INSTANCE_DEACTIVATED || event.type() == Type.INSTANCE_REMOVED) {
+ scheduleStaleLeadershipPurge(0);
+ }
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/EventuallyConsistentMapBuilderImpl.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/EventuallyConsistentMapBuilderImpl.java
new file mode 100644
index 0000000..7ab30ea
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/EventuallyConsistentMapBuilderImpl.java
@@ -0,0 +1,168 @@
+/*
+ * 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.KryoNamespace;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.persistence.PersistenceService;
+import org.onosproject.store.Timestamp;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.EventuallyConsistentMapBuilder;
+
+import java.util.Collection;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Eventually consistent map builder.
+ */
+public class EventuallyConsistentMapBuilderImpl<K, V>
+ implements EventuallyConsistentMapBuilder<K, V> {
+ private final ClusterService clusterService;
+ private final ClusterCommunicationService clusterCommunicator;
+
+ private String name;
+ private KryoNamespace.Builder serializerBuilder;
+ private ExecutorService eventExecutor;
+ private ExecutorService communicationExecutor;
+ private ScheduledExecutorService backgroundExecutor;
+ private BiFunction<K, V, Timestamp> timestampProvider;
+ private BiFunction<K, V, Collection<NodeId>> peerUpdateFunction;
+ private boolean tombstonesDisabled = false;
+ private long antiEntropyPeriod = 5;
+ private TimeUnit antiEntropyTimeUnit = TimeUnit.SECONDS;
+ private boolean convergeFaster = false;
+ private boolean persistent = false;
+ private boolean persistentMap = false;
+ private final PersistenceService persistenceService;
+
+ /**
+ * Creates a new eventually consistent map builder.
+ *
+ * @param clusterService cluster service
+ * @param clusterCommunicator cluster communication service
+ */
+ public EventuallyConsistentMapBuilderImpl(ClusterService clusterService,
+ ClusterCommunicationService clusterCommunicator,
+ PersistenceService persistenceService) {
+ this.persistenceService = persistenceService;
+ this.clusterService = checkNotNull(clusterService);
+ this.clusterCommunicator = checkNotNull(clusterCommunicator);
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withName(String name) {
+ this.name = checkNotNull(name);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withSerializer(
+ KryoNamespace.Builder serializerBuilder) {
+ this.serializerBuilder = checkNotNull(serializerBuilder);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withTimestampProvider(
+ BiFunction<K, V, Timestamp> timestampProvider) {
+ this.timestampProvider = checkNotNull(timestampProvider);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withEventExecutor(ExecutorService executor) {
+ this.eventExecutor = checkNotNull(executor);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withCommunicationExecutor(
+ ExecutorService executor) {
+ communicationExecutor = checkNotNull(executor);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withBackgroundExecutor(ScheduledExecutorService executor) {
+ this.backgroundExecutor = checkNotNull(executor);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withPeerUpdateFunction(
+ BiFunction<K, V, Collection<NodeId>> peerUpdateFunction) {
+ this.peerUpdateFunction = checkNotNull(peerUpdateFunction);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withTombstonesDisabled() {
+ tombstonesDisabled = true;
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withAntiEntropyPeriod(long period, TimeUnit unit) {
+ checkArgument(period > 0, "anti-entropy period must be greater than 0");
+ antiEntropyPeriod = period;
+ antiEntropyTimeUnit = checkNotNull(unit);
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withFasterConvergence() {
+ convergeFaster = true;
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMapBuilder<K, V> withPersistence() {
+ checkNotNull(this.persistenceService);
+ persistent = true;
+ return this;
+ }
+
+ @Override
+ public EventuallyConsistentMap<K, V> build() {
+ checkNotNull(name, "name is a mandatory parameter");
+ checkNotNull(serializerBuilder, "serializerBuilder is a mandatory parameter");
+ checkNotNull(timestampProvider, "timestampProvider is a mandatory parameter");
+
+ return new EventuallyConsistentMapImpl<>(name,
+ clusterService,
+ clusterCommunicator,
+ serializerBuilder,
+ timestampProvider,
+ peerUpdateFunction,
+ eventExecutor,
+ communicationExecutor,
+ backgroundExecutor,
+ tombstonesDisabled,
+ antiEntropyPeriod,
+ antiEntropyTimeUnit,
+ convergeFaster,
+ persistent,
+ persistenceService);
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/EventuallyConsistentMapImpl.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/EventuallyConsistentMapImpl.java
new file mode 100644
index 0000000..f595e6a
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/EventuallyConsistentMapImpl.java
@@ -0,0 +1,685 @@
+/*
+ * 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 com.google.common.base.Preconditions.checkState;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.BoundedThreadPool.newFixedThreadPool;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT;
+import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.REMOVE;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.Timer;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.onlab.util.AbstractAccumulator;
+import org.onlab.util.KryoNamespace;
+import org.onlab.util.SlidingWindowCounter;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.persistence.PersistenceService;
+import org.onosproject.store.LogicalTimestamp;
+import org.onosproject.store.Timestamp;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.cluster.messaging.MessageSubject;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.serializers.KryoSerializer;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.EventuallyConsistentMapEvent;
+import org.onosproject.store.service.EventuallyConsistentMapListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.WallClockTimestamp;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Collections2;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+/**
+ * Distributed Map implementation which uses optimistic replication and gossip
+ * based techniques to provide an eventually consistent data store.
+ */
+public class EventuallyConsistentMapImpl<K, V>
+ implements EventuallyConsistentMap<K, V> {
+
+ private static final Logger log = LoggerFactory.getLogger(EventuallyConsistentMapImpl.class);
+
+ private final Map<K, MapValue<V>> items;
+
+ private final ClusterService clusterService;
+ private final ClusterCommunicationService clusterCommunicator;
+ private final KryoSerializer serializer;
+ private final NodeId localNodeId;
+ private final PersistenceService persistenceService;
+
+ private final BiFunction<K, V, Timestamp> timestampProvider;
+
+ private final MessageSubject updateMessageSubject;
+ private final MessageSubject antiEntropyAdvertisementSubject;
+
+ private final Set<EventuallyConsistentMapListener<K, V>> listeners
+ = Sets.newCopyOnWriteArraySet();
+
+ private final ExecutorService executor;
+ private final ScheduledExecutorService backgroundExecutor;
+ private final BiFunction<K, V, Collection<NodeId>> peerUpdateFunction;
+
+ private final ExecutorService communicationExecutor;
+ private final Map<NodeId, EventAccumulator> senderPending;
+
+ private final String mapName;
+
+ private volatile boolean destroyed = false;
+ private static final String ERROR_DESTROYED = " map is already destroyed";
+ private final String destroyedMessage;
+
+ private static final String ERROR_NULL_KEY = "Key cannot be null";
+ private static final String ERROR_NULL_VALUE = "Null values are not allowed";
+
+ private final long initialDelaySec = 5;
+ private final boolean lightweightAntiEntropy;
+ private final boolean tombstonesDisabled;
+
+ private static final int WINDOW_SIZE = 5;
+ private static final int HIGH_LOAD_THRESHOLD = 0;
+ private static final int LOAD_WINDOW = 2;
+ private SlidingWindowCounter counter = new SlidingWindowCounter(WINDOW_SIZE);
+
+ private final boolean persistent;
+
+ private static final String PERSISTENT_LOCAL_MAP_NAME = "itemsMap";
+
+
+ /**
+ * Creates a new eventually consistent map shared amongst multiple instances.
+ * <p>
+ * See {@link org.onosproject.store.service.EventuallyConsistentMapBuilder}
+ * for more description of the parameters expected by the map.
+ * </p>
+ *
+ * @param mapName a String identifier for the map.
+ * @param clusterService the cluster service
+ * @param clusterCommunicator the cluster communications service
+ * @param serializerBuilder a Kryo namespace builder that can serialize
+ * both K and V
+ * @param timestampProvider provider of timestamps for K and V
+ * @param peerUpdateFunction function that provides a set of nodes to immediately
+ * update to when there writes to the map
+ * @param eventExecutor executor to use for processing incoming
+ * events from peers
+ * @param communicationExecutor executor to use for sending events to peers
+ * @param backgroundExecutor executor to use for background anti-entropy
+ * tasks
+ * @param tombstonesDisabled true if this map should not maintain
+ * tombstones
+ * @param antiEntropyPeriod period that the anti-entropy task should run
+ * @param antiEntropyTimeUnit time unit for anti-entropy period
+ * @param convergeFaster make anti-entropy try to converge faster
+ * @param persistent persist data to disk
+ */
+ EventuallyConsistentMapImpl(String mapName,
+ ClusterService clusterService,
+ ClusterCommunicationService clusterCommunicator,
+ KryoNamespace.Builder serializerBuilder,
+ BiFunction<K, V, Timestamp> timestampProvider,
+ BiFunction<K, V, Collection<NodeId>> peerUpdateFunction,
+ ExecutorService eventExecutor,
+ ExecutorService communicationExecutor,
+ ScheduledExecutorService backgroundExecutor,
+ boolean tombstonesDisabled,
+ long antiEntropyPeriod,
+ TimeUnit antiEntropyTimeUnit,
+ boolean convergeFaster,
+ boolean persistent,
+ PersistenceService persistenceService) {
+ this.mapName = mapName;
+ this.serializer = createSerializer(serializerBuilder);
+ this.persistenceService = persistenceService;
+ this.persistent =
+ persistent;
+ if (persistent) {
+ items = this.persistenceService.<K, MapValue<V>>persistentMapBuilder()
+ .withName(PERSISTENT_LOCAL_MAP_NAME)
+ .withSerializer(new Serializer() {
+
+ @Override
+ public <T> byte[] encode(T object) {
+ return EventuallyConsistentMapImpl.this.serializer.encode(object);
+ }
+
+ @Override
+ public <T> T decode(byte[] bytes) {
+ return EventuallyConsistentMapImpl.this.serializer.decode(bytes);
+ }
+ })
+ .build();
+ } else {
+ items = Maps.newConcurrentMap();
+ }
+ senderPending = Maps.newConcurrentMap();
+ destroyedMessage = mapName + ERROR_DESTROYED;
+
+ this.clusterService = clusterService;
+ this.clusterCommunicator = clusterCommunicator;
+ this.localNodeId = clusterService.getLocalNode().id();
+
+ this.timestampProvider = timestampProvider;
+
+ if (peerUpdateFunction != null) {
+ this.peerUpdateFunction = peerUpdateFunction;
+ } else {
+ this.peerUpdateFunction = (key, value) -> clusterService.getNodes().stream()
+ .map(ControllerNode::id)
+ .filter(nodeId -> !nodeId.equals(localNodeId))
+ .collect(Collectors.toList());
+ }
+
+ if (eventExecutor != null) {
+ this.executor = eventExecutor;
+ } else {
+ // should be a normal executor; it's used for receiving messages
+ this.executor =
+ Executors.newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-fg-%d"));
+ }
+
+ if (communicationExecutor != null) {
+ this.communicationExecutor = communicationExecutor;
+ } else {
+ // sending executor; should be capped
+ //TODO this probably doesn't need to be bounded anymore
+ this.communicationExecutor =
+ newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-publish-%d"));
+ }
+
+
+ if (backgroundExecutor != null) {
+ this.backgroundExecutor = backgroundExecutor;
+ } else {
+ this.backgroundExecutor =
+ newSingleThreadScheduledExecutor(groupedThreads("onos/ecm", mapName + "-bg-%d"));
+ }
+
+ // start anti-entropy thread
+ this.backgroundExecutor.scheduleAtFixedRate(this::sendAdvertisement,
+ initialDelaySec, antiEntropyPeriod,
+ antiEntropyTimeUnit);
+
+ updateMessageSubject = new MessageSubject("ecm-" + mapName + "-update");
+ clusterCommunicator.addSubscriber(updateMessageSubject,
+ serializer::decode,
+ this::processUpdates,
+ this.executor);
+
+ antiEntropyAdvertisementSubject = new MessageSubject("ecm-" + mapName + "-anti-entropy");
+ clusterCommunicator.addSubscriber(antiEntropyAdvertisementSubject,
+ serializer::decode,
+ this::handleAntiEntropyAdvertisement,
+ this.backgroundExecutor);
+
+ this.tombstonesDisabled = tombstonesDisabled;
+ this.lightweightAntiEntropy = !convergeFaster;
+ }
+
+ private KryoSerializer createSerializer(KryoNamespace.Builder builder) {
+ return new KryoSerializer() {
+ @Override
+ protected void setupKryoPool() {
+ // Add the map's internal helper classes to the user-supplied serializer
+ serializerPool = builder
+ .register(KryoNamespaces.BASIC)
+ .nextId(KryoNamespaces.BEGIN_USER_CUSTOM_ID)
+ .register(LogicalTimestamp.class)
+ .register(WallClockTimestamp.class)
+ .register(AntiEntropyAdvertisement.class)
+ .register(UpdateEntry.class)
+ .register(MapValue.class)
+ .register(MapValue.Digest.class)
+ .build();
+ }
+ };
+ }
+
+ @Override
+ public String name() {
+ return mapName;
+ }
+
+ @Override
+ public int size() {
+ checkState(!destroyed, destroyedMessage);
+ // TODO: Maintain a separate counter for tracking live elements in map.
+ return Maps.filterValues(items, MapValue::isAlive).size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ checkState(!destroyed, destroyedMessage);
+ return size() == 0;
+ }
+
+ @Override
+ public boolean containsKey(K key) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ return get(key) != null;
+ }
+
+ @Override
+ public boolean containsValue(V value) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(value, ERROR_NULL_VALUE);
+ return items.values()
+ .stream()
+ .filter(MapValue::isAlive)
+ .anyMatch(v -> value.equals(v.get()));
+ }
+
+ @Override
+ public V get(K key) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+
+ MapValue<V> value = items.get(key);
+ return (value == null || value.isTombstone()) ? null : value.get();
+ }
+
+ @Override
+ public void put(K key, V value) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ checkNotNull(value, ERROR_NULL_VALUE);
+
+ MapValue<V> newValue = new MapValue<>(value, timestampProvider.apply(key, value));
+ if (putInternal(key, newValue)) {
+ notifyPeers(new UpdateEntry<>(key, newValue), peerUpdateFunction.apply(key, value));
+ notifyListeners(new EventuallyConsistentMapEvent<>(mapName, PUT, key, value));
+ }
+ }
+
+ @Override
+ public V remove(K key) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ return removeAndNotify(key, null);
+ }
+
+ @Override
+ public void remove(K key, V value) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ checkNotNull(value, ERROR_NULL_VALUE);
+ removeAndNotify(key, value);
+ }
+
+ private V removeAndNotify(K key, V value) {
+ Timestamp timestamp = timestampProvider.apply(key, value);
+ Optional<MapValue<V>> tombstone = tombstonesDisabled || timestamp == null
+ ? Optional.empty() : Optional.of(MapValue.tombstone(timestamp));
+ MapValue<V> previousValue = removeInternal(key, Optional.ofNullable(value), tombstone);
+ if (previousValue != null) {
+ notifyPeers(new UpdateEntry<>(key, tombstone.orElse(null)),
+ peerUpdateFunction.apply(key, previousValue.get()));
+ if (previousValue.isAlive()) {
+ notifyListeners(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get()));
+ }
+ }
+ return previousValue != null ? previousValue.get() : null;
+ }
+
+ private MapValue<V> removeInternal(K key, Optional<V> value, Optional<MapValue<V>> tombstone) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ checkNotNull(value, ERROR_NULL_VALUE);
+ tombstone.ifPresent(v -> checkState(v.isTombstone()));
+
+ counter.incrementCount();
+ AtomicBoolean updated = new AtomicBoolean(false);
+ AtomicReference<MapValue<V>> previousValue = new AtomicReference<>();
+ items.compute(key, (k, existing) -> {
+ boolean valueMatches = true;
+ if (value.isPresent() && existing != null && existing.isAlive()) {
+ valueMatches = Objects.equals(value.get(), existing.get());
+ }
+ if (existing == null) {
+ log.trace("ECMap Remove: Existing value for key {} is already null", k);
+ }
+ if (valueMatches) {
+ if (existing == null) {
+ updated.set(tombstone.isPresent());
+ } else {
+ updated.set(!tombstone.isPresent() || tombstone.get().isNewerThan(existing));
+ }
+ }
+ if (updated.get()) {
+ previousValue.set(existing);
+ return tombstone.orElse(null);
+ } else {
+ return existing;
+ }
+ });
+ return previousValue.get();
+ }
+
+ @Override
+ public V compute(K key, BiFunction<K, V, V> recomputeFunction) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ checkNotNull(recomputeFunction, "Recompute function cannot be null");
+
+ AtomicBoolean updated = new AtomicBoolean(false);
+ AtomicReference<MapValue<V>> previousValue = new AtomicReference<>();
+ MapValue<V> computedValue = items.compute(key, (k, mv) -> {
+ previousValue.set(mv);
+ V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get());
+ MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue));
+ if (mv == null || newValue.isNewerThan(mv)) {
+ updated.set(true);
+ return newValue;
+ } else {
+ return mv;
+ }
+ });
+ if (updated.get()) {
+ notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get()));
+ EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT;
+ V value = computedValue.isTombstone()
+ ? previousValue.get() == null ? null : previousValue.get().get()
+ : computedValue.get();
+ if (value != null) {
+ notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value));
+ }
+ }
+ return computedValue.get();
+ }
+
+ @Override
+ public void putAll(Map<? extends K, ? extends V> m) {
+ checkState(!destroyed, destroyedMessage);
+ m.forEach(this::put);
+ }
+
+ @Override
+ public void clear() {
+ checkState(!destroyed, destroyedMessage);
+ Maps.filterValues(items, MapValue::isAlive)
+ .forEach((k, v) -> remove(k));
+ }
+
+ @Override
+ public Set<K> keySet() {
+ checkState(!destroyed, destroyedMessage);
+ return Maps.filterValues(items, MapValue::isAlive)
+ .keySet();
+ }
+
+ @Override
+ public Collection<V> values() {
+ checkState(!destroyed, destroyedMessage);
+ return Collections2.transform(Maps.filterValues(items, MapValue::isAlive).values(), MapValue::get);
+ }
+
+ @Override
+ public Set<Map.Entry<K, V>> entrySet() {
+ checkState(!destroyed, destroyedMessage);
+ return Maps.filterValues(items, MapValue::isAlive)
+ .entrySet()
+ .stream()
+ .map(e -> Pair.of(e.getKey(), e.getValue().get()))
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Returns true if newValue was accepted i.e. map is updated.
+ *
+ * @param key key
+ * @param newValue proposed new value
+ * @return true if update happened; false if map already contains a more recent value for the key
+ */
+ private boolean putInternal(K key, MapValue<V> newValue) {
+ checkState(!destroyed, destroyedMessage);
+ checkNotNull(key, ERROR_NULL_KEY);
+ checkNotNull(newValue, ERROR_NULL_VALUE);
+ checkState(newValue.isAlive());
+ counter.incrementCount();
+ AtomicBoolean updated = new AtomicBoolean(false);
+ items.compute(key, (k, existing) -> {
+ if (existing == null || newValue.isNewerThan(existing)) {
+ updated.set(true);
+ return newValue;
+ }
+ return existing;
+ });
+ return updated.get();
+ }
+
+ @Override
+ public void addListener(EventuallyConsistentMapListener<K, V> listener) {
+ checkState(!destroyed, destroyedMessage);
+
+ listeners.add(checkNotNull(listener));
+ }
+
+ @Override
+ public void removeListener(EventuallyConsistentMapListener<K, V> listener) {
+ checkState(!destroyed, destroyedMessage);
+
+ listeners.remove(checkNotNull(listener));
+ }
+
+ @Override
+ public void destroy() {
+ destroyed = true;
+
+ executor.shutdown();
+ backgroundExecutor.shutdown();
+ communicationExecutor.shutdown();
+
+ listeners.clear();
+
+ clusterCommunicator.removeSubscriber(updateMessageSubject);
+ clusterCommunicator.removeSubscriber(antiEntropyAdvertisementSubject);
+ }
+
+ private void notifyListeners(EventuallyConsistentMapEvent<K, V> event) {
+ listeners.forEach(listener -> listener.event(event));
+ }
+
+ private void notifyPeers(UpdateEntry<K, V> event, Collection<NodeId> peers) {
+ queueUpdate(event, peers);
+ }
+
+ private void queueUpdate(UpdateEntry<K, V> event, Collection<NodeId> peers) {
+ if (peers == null) {
+ // we have no friends :(
+ return;
+ }
+ peers.forEach(node ->
+ senderPending.computeIfAbsent(node, unusedKey -> new EventAccumulator(node)).add(event)
+ );
+ }
+
+ private boolean underHighLoad() {
+ return counter.get(LOAD_WINDOW) > HIGH_LOAD_THRESHOLD;
+ }
+
+ private void sendAdvertisement() {
+ try {
+ if (underHighLoad() || destroyed) {
+ return;
+ }
+ pickRandomActivePeer().ifPresent(this::sendAdvertisementToPeer);
+ } catch (Exception e) {
+ // Catch all exceptions to avoid scheduled task being suppressed.
+ log.error("Exception thrown while sending advertisement", e);
+ }
+ }
+
+ private Optional<NodeId> pickRandomActivePeer() {
+ List<NodeId> activePeers = clusterService.getNodes()
+ .stream()
+ .map(ControllerNode::id)
+ .filter(id -> !localNodeId.equals(id))
+ .filter(id -> clusterService.getState(id) == ControllerNode.State.ACTIVE)
+ .collect(Collectors.toList());
+ Collections.shuffle(activePeers);
+ return activePeers.isEmpty() ? Optional.empty() : Optional.of(activePeers.get(0));
+ }
+
+ private void sendAdvertisementToPeer(NodeId peer) {
+ clusterCommunicator.unicast(createAdvertisement(),
+ antiEntropyAdvertisementSubject,
+ serializer::encode,
+ peer)
+ .whenComplete((result, error) -> {
+ if (error != null) {
+ log.debug("Failed to send anti-entropy advertisement to {}", peer, error);
+ }
+ });
+ }
+
+ private AntiEntropyAdvertisement<K> createAdvertisement() {
+ return new AntiEntropyAdvertisement<K>(localNodeId,
+ ImmutableMap.copyOf(Maps.transformValues(items, MapValue::digest)));
+ }
+
+ private void handleAntiEntropyAdvertisement(AntiEntropyAdvertisement<K> ad) {
+ if (destroyed || underHighLoad()) {
+ return;
+ }
+ try {
+ if (log.isTraceEnabled()) {
+ log.trace("Received anti-entropy advertisement from {} for {} with {} entries in it",
+ mapName, ad.sender(), ad.digest().size());
+ }
+ antiEntropyCheckLocalItems(ad).forEach(this::notifyListeners);
+
+ if (!lightweightAntiEntropy) {
+ // if remote ad has any entries that the local copy is missing, actively sync
+ // TODO: Missing keys is not the way local copy can be behind.
+ if (Sets.difference(ad.digest().keySet(), items.keySet()).size() > 0) {
+ // TODO: Send ad for missing keys and for entries that are stale
+ sendAdvertisementToPeer(ad.sender());
+ }
+ }
+ } catch (Exception e) {
+ log.warn("Error handling anti-entropy advertisement", e);
+ }
+ }
+
+ /**
+ * Processes anti-entropy ad from peer by taking following actions:
+ * 1. If peer has an old entry, updates peer.
+ * 2. If peer indicates an entry is removed and has a more recent
+ * timestamp than the local entry, update local state.
+ */
+ private List<EventuallyConsistentMapEvent<K, V>> antiEntropyCheckLocalItems(
+ AntiEntropyAdvertisement<K> ad) {
+ final List<EventuallyConsistentMapEvent<K, V>> externalEvents = Lists.newLinkedList();
+ final NodeId sender = ad.sender();
+ items.forEach((key, localValue) -> {
+ MapValue.Digest remoteValueDigest = ad.digest().get(key);
+ if (remoteValueDigest == null || localValue.isNewerThan(remoteValueDigest.timestamp())) {
+ // local value is more recent, push to sender
+ queueUpdate(new UpdateEntry<>(key, localValue), ImmutableList.of(sender));
+ }
+ if (remoteValueDigest != null
+ && remoteValueDigest.isNewerThan(localValue.digest())
+ && remoteValueDigest.isTombstone()) {
+ MapValue<V> tombstone = MapValue.tombstone(remoteValueDigest.timestamp());
+ MapValue<V> previousValue = removeInternal(key,
+ Optional.empty(),
+ Optional.of(tombstone));
+ if (previousValue != null && previousValue.isAlive()) {
+ externalEvents.add(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get()));
+ }
+ }
+ });
+ return externalEvents;
+ }
+
+ private void processUpdates(Collection<UpdateEntry<K, V>> updates) {
+ if (destroyed) {
+ return;
+ }
+ updates.forEach(update -> {
+ final K key = update.key();
+ final MapValue<V> value = update.value();
+ if (value == null || value.isTombstone()) {
+ MapValue<V> previousValue = removeInternal(key, Optional.empty(), Optional.ofNullable(value));
+ if (previousValue != null && previousValue.isAlive()) {
+ notifyListeners(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get()));
+ }
+ } else if (putInternal(key, value)) {
+ notifyListeners(new EventuallyConsistentMapEvent<>(mapName, PUT, key, value.get()));
+ }
+ });
+ }
+
+ // TODO pull this into the class if this gets pulled out...
+ private static final int DEFAULT_MAX_EVENTS = 1000;
+ private static final int DEFAULT_MAX_IDLE_MS = 10;
+ private static final int DEFAULT_MAX_BATCH_MS = 50;
+ private static final Timer TIMER = new Timer("onos-ecm-sender-events");
+
+ private final class EventAccumulator extends AbstractAccumulator<UpdateEntry<K, V>> {
+
+ private final NodeId peer;
+
+ private EventAccumulator(NodeId peer) {
+ super(TIMER, DEFAULT_MAX_EVENTS, DEFAULT_MAX_BATCH_MS, DEFAULT_MAX_IDLE_MS);
+ this.peer = peer;
+ }
+
+ @Override
+ public void processItems(List<UpdateEntry<K, V>> items) {
+ Map<K, UpdateEntry<K, V>> map = Maps.newHashMap();
+ items.forEach(item -> map.compute(item.key(), (key, existing) ->
+ item.isNewerThan(existing) ? item : existing));
+ communicationExecutor.submit(() -> {
+ clusterCommunicator.unicast(ImmutableList.copyOf(map.values()),
+ updateMessageSubject,
+ serializer::encode,
+ peer)
+ .whenComplete((result, error) -> {
+ if (error != null) {
+ log.debug("Failed to send to {}", peer, error);
+ }
+ });
+ });
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MapDbPersistentStore.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MapDbPersistentStore.java
new file mode 100644
index 0000000..2a58992
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MapDbPersistentStore.java
@@ -0,0 +1,103 @@
+/*
+ * 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.mapdb.DB;
+import org.mapdb.DBMaker;
+import org.mapdb.Hasher;
+import org.mapdb.Serializer;
+import org.onosproject.store.serializers.KryoSerializer;
+
+import java.io.File;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * MapDB based implementation of a persistent store.
+ */
+class MapDbPersistentStore<K, V> implements PersistentStore<K, V> {
+
+ private final ExecutorService executor;
+ private final KryoSerializer serializer;
+
+ private final DB database;
+
+ private final Map<byte[], byte[]> items;
+
+ /**
+ * Creates a new MapDB based persistent store.
+ *
+ * @param filename filename of the database on disk
+ * @param executor executor to use for tasks that write to the disk
+ * @param serializer serializer for keys and values
+ */
+ MapDbPersistentStore(String filename, ExecutorService executor,
+ KryoSerializer serializer) {
+ this.executor = checkNotNull(executor);
+ this.serializer = checkNotNull(serializer);
+
+ File databaseFile = new File(filename);
+
+ database = DBMaker.newFileDB(databaseFile).make();
+
+ items = database.createHashMap("items")
+ .keySerializer(Serializer.BYTE_ARRAY)
+ .valueSerializer(Serializer.BYTE_ARRAY)
+ .hasher(Hasher.BYTE_ARRAY)
+ .makeOrGet();
+ }
+
+ @Override
+ public void readInto(Map<K, MapValue<V>> items) {
+ this.items.forEach((keyBytes, valueBytes) ->
+ items.put(serializer.decode(keyBytes),
+ serializer.decode(valueBytes)));
+ }
+
+ @Override
+ public void update(K key, MapValue<V> value) {
+ executor.submit(() -> updateInternal(key, value));
+ }
+
+ @Override
+ public void remove(K key) {
+ executor.submit(() -> removeInternal(key));
+ }
+
+ private void updateInternal(K key, MapValue<V> newValue) {
+ byte[] keyBytes = serializer.encode(key);
+
+ items.compute(keyBytes, (k, existingBytes) -> {
+ MapValue<V> existing = existingBytes == null ? null :
+ serializer.decode(existingBytes);
+ if (existing == null || newValue.isNewerThan(existing)) {
+ return serializer.encode(newValue);
+ } else {
+ return existingBytes;
+ }
+ });
+ database.commit();
+ }
+
+ private void removeInternal(K key) {
+ byte[] keyBytes = serializer.encode(key);
+ items.remove(keyBytes);
+ database.commit();
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MapValue.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MapValue.java
new file mode 100644
index 0000000..8d98a23
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MapValue.java
@@ -0,0 +1,163 @@
+/*
+ * 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.Timestamp;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Representation of a value in EventuallyConsistentMap.
+ *
+ * @param <V> value type
+ */
+public class MapValue<V> implements Comparable<MapValue<V>> {
+ private final Timestamp timestamp;
+ private final V value;
+
+ /**
+ * Creates a tombstone value with the specified timestamp.
+ * @param timestamp timestamp for tombstone
+ * @return tombstone MapValue
+ *
+ * @param <U> value type
+ */
+ public static <U> MapValue<U> tombstone(Timestamp timestamp) {
+ return new MapValue<>(null, timestamp);
+ }
+
+ public MapValue(V value, Timestamp timestamp) {
+ this.value = value;
+ this.timestamp = checkNotNull(timestamp, "Timestamp cannot be null");
+ }
+
+ public boolean isTombstone() {
+ return value == null;
+ }
+
+ public boolean isAlive() {
+ return value != null;
+ }
+
+ public Timestamp timestamp() {
+ return timestamp;
+ }
+
+ public V get() {
+ return value;
+ }
+
+ @Override
+ public int compareTo(MapValue<V> o) {
+ return this.timestamp.compareTo(o.timestamp);
+ }
+
+ public boolean isNewerThan(MapValue<V> other) {
+ if (other == null) {
+ return true;
+ }
+ return this.timestamp.isNewerThan(other.timestamp);
+ }
+
+ public boolean isNewerThan(Timestamp timestamp) {
+ return this.timestamp.isNewerThan(timestamp);
+ }
+
+ public Digest digest() {
+ return new Digest(timestamp, isTombstone());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(timestamp, value);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean equals(Object other) {
+ if (other instanceof MapValue) {
+ MapValue<V> that = (MapValue<V>) other;
+ return Objects.equal(this.timestamp, that.timestamp) &&
+ Objects.equal(this.value, that.value);
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("timestamp", timestamp)
+ .add("value", value)
+ .toString();
+ }
+
+ @SuppressWarnings("unused")
+ private MapValue() {
+ this.timestamp = null;
+ this.value = null;
+ }
+
+ /**
+ * Digest or summary of a MapValue for use during Anti-Entropy exchanges.
+ */
+ public static class Digest {
+ private final Timestamp timestamp;
+ private final boolean isTombstone;
+
+ public Digest(Timestamp timestamp, boolean isTombstone) {
+ this.timestamp = timestamp;
+ this.isTombstone = isTombstone;
+ }
+
+ public Timestamp timestamp() {
+ return timestamp;
+ }
+
+ public boolean isTombstone() {
+ return isTombstone;
+ }
+
+ public boolean isNewerThan(Digest other) {
+ return timestamp.isNewerThan(other.timestamp);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(timestamp, isTombstone);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other instanceof Digest) {
+ Digest that = (Digest) other;
+ return Objects.equal(this.timestamp, that.timestamp) &&
+ Objects.equal(this.isTombstone, that.isTombstone);
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("timestamp", timestamp)
+ .add("isTombstone", isTombstone)
+ .toString();
+ }
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MappingSet.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MappingSet.java
new file mode 100644
index 0000000..6ca4d4f
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MappingSet.java
@@ -0,0 +1,131 @@
+/*
+ * 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();
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteringAgent.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteringAgent.java
new file mode 100644
index 0000000..27a422e
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MeteringAgent.java
@@ -0,0 +1,134 @@
+/*
+ * 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.codahale.metrics.Counter;
+import com.codahale.metrics.Timer;
+import com.google.common.collect.Maps;
+import org.onlab.metrics.MetricsComponent;
+import org.onlab.metrics.MetricsFeature;
+import org.onlab.metrics.MetricsService;
+import org.onlab.osgi.DefaultServiceDirectory;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Agent that implements usage and performance monitoring via the metrics service.
+ */
+public class MeteringAgent {
+
+ private Counter exceptionCounter;
+ private Counter perObjExceptionCounter;
+ private MetricsService metricsService;
+ private MetricsComponent metricsComponent;
+ private MetricsFeature metricsFeature;
+ private final Map<String, Timer> perObjOpTimers = Maps.newConcurrentMap();
+ private final Map<String, Timer> perOpTimers = Maps.newConcurrentMap();
+ private Timer perPrimitiveTimer;
+ private Timer perObjTimer;
+ private MetricsFeature wildcard;
+ private final boolean activated;
+ private Context nullTimer;
+
+ /**
+ * Constructs a new MeteringAgent for a given distributed primitive.
+ * Instantiates the metrics service
+ * Initializes all the general metrics for that object
+ *
+ * @param primitiveName Type of primitive to be metered
+ * @param objName Global name of the primitive
+ * @param activated boolean flag for whether metering is enabled or not
+ */
+ public MeteringAgent(String primitiveName, String objName, boolean activated) {
+ checkNotNull(objName, "Object name cannot be null");
+ this.activated = activated;
+ nullTimer = new Context(null, "");
+ if (this.activated) {
+ this.metricsService = DefaultServiceDirectory.getService(MetricsService.class);
+ this.metricsComponent = metricsService.registerComponent(primitiveName);
+ this.metricsFeature = metricsComponent.registerFeature(objName);
+ this.wildcard = metricsComponent.registerFeature("*");
+ this.perObjTimer = metricsService.createTimer(metricsComponent, metricsFeature, "*");
+ this.perPrimitiveTimer = metricsService.createTimer(metricsComponent, wildcard, "*");
+ this.perObjExceptionCounter = metricsService.createCounter(metricsComponent, metricsFeature, "exceptions");
+ this.exceptionCounter = metricsService.createCounter(metricsComponent, wildcard, "exceptions");
+ }
+ }
+
+ /**
+ * Initializes a specific timer for a given operation.
+ *
+ * @param op Specific operation being metered
+ * @return timer context
+ */
+ public Context startTimer(String op) {
+ if (!activated) {
+ return nullTimer;
+ }
+ // Check if timer exists, if it doesn't creates it
+ final Timer currTimer = perObjOpTimers.computeIfAbsent(op, timer ->
+ metricsService.createTimer(metricsComponent, metricsFeature, op));
+ perOpTimers.computeIfAbsent(op, timer -> metricsService.createTimer(metricsComponent, wildcard, op));
+ // Starts timer
+ return new Context(currTimer.time(), op);
+ }
+
+ /**
+ * Timer.Context with a specific operation.
+ */
+ public class Context {
+ private final Timer.Context context;
+ private final String operation;
+
+ /**
+ * Constructs Context.
+ *
+ * @param context context
+ * @param operation operation name
+ */
+ public Context(Timer.Context context, String operation) {
+ this.context = context;
+ this.operation = operation;
+ }
+
+ /**
+ * Stops timer given a specific context and updates all related metrics.
+ * @param e throwable
+ */
+ public void stop(Throwable e) {
+ if (!activated) {
+ return;
+ }
+ if (e == null) {
+ //Stop and updates timer with specific measurements per map, per operation
+ final long time = context.stop();
+ //updates timer with aggregated measurements per map
+ perOpTimers.get(operation).update(time, TimeUnit.NANOSECONDS);
+ //updates timer with aggregated measurements per map
+ perObjTimer.update(time, TimeUnit.NANOSECONDS);
+ //updates timer with aggregated measurements per all Consistent Maps
+ perPrimitiveTimer.update(time, TimeUnit.NANOSECONDS);
+ } else {
+ exceptionCounter.inc();
+ perObjExceptionCounter.inc();
+ }
+ }
+ }
+
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MutexExecutionManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MutexExecutionManager.java
new file mode 100644
index 0000000..5946fdb
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/MutexExecutionManager.java
@@ -0,0 +1,315 @@
+/*
+ * 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.slf4j.LoggerFactory.getLogger;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+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.Service;
+import org.onlab.util.Tools;
+import org.onosproject.cluster.ClusterEvent;
+import org.onosproject.cluster.ClusterEventListener;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.ControllerNode.State;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.ConsistentMapException;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.MutexExecutionService;
+import org.onosproject.store.service.MutexTask;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * Implementation of a MutexExecutionService.
+ */
+@Component(immediate = true)
+@Service
+public class MutexExecutionManager implements MutexExecutionService {
+
+ private final Logger log = getLogger(getClass());
+
+ protected ConsistentMap<String, MutexState> lockMap;
+ protected NodeId localNodeId;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StorageService storageService;
+
+ private final MapEventListener<String, MutexState> mapEventListener = new InternalLockMapEventListener();
+ private final ClusterEventListener clusterEventListener = new InternalClusterEventListener();
+
+ private Map<String, CompletableFuture<MutexState>> pending = Maps.newConcurrentMap();
+ private Map<String, InnerMutexTask> activeTasks = Maps.newConcurrentMap();
+
+ @Activate
+ public void activate() {
+ localNodeId = clusterService.getLocalNode().id();
+ lockMap = storageService.<String, MutexState>consistentMapBuilder()
+ .withName("onos-mutexes")
+ .withSerializer(Serializer.using(Arrays.asList(KryoNamespaces.API), MutexState.class))
+ .withPartitionsDisabled()
+ .build();
+ lockMap.addListener(mapEventListener);
+ clusterService.addListener(clusterEventListener);
+ releaseOldLocks();
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ lockMap.removeListener(mapEventListener);
+ pending.values().forEach(future -> future.cancel(true));
+ activeTasks.forEach((k, v) -> {
+ v.stop();
+ unlock(k);
+ });
+ clusterService.removeListener(clusterEventListener);
+ log.info("Stopped");
+ }
+
+ @Override
+ public CompletableFuture<Void> execute(MutexTask task, String exclusionPath, Executor executor) {
+ return lock(exclusionPath)
+ .thenApply(state -> activeTasks.computeIfAbsent(exclusionPath,
+ k -> new InnerMutexTask(exclusionPath,
+ task,
+ state.term())))
+ .thenAcceptAsync(t -> t.start(), executor)
+ .whenComplete((r, e) -> unlock(exclusionPath));
+ }
+
+ protected CompletableFuture<MutexState> lock(String exclusionPath) {
+ CompletableFuture<MutexState> future =
+ pending.computeIfAbsent(exclusionPath, k -> new CompletableFuture<>());
+ tryLock(exclusionPath);
+ return future;
+ }
+
+ /**
+ * Attempts to acquire lock for a path. If lock is held by some other node, adds this node to
+ * the wait list.
+ * @param exclusionPath exclusion path
+ */
+ protected void tryLock(String exclusionPath) {
+ Tools.retryable(() -> lockMap.asJavaMap()
+ .compute(exclusionPath,
+ (k, v) -> MutexState.admit(v, localNodeId)),
+ ConsistentMapException.ConcurrentModification.class,
+ Integer.MAX_VALUE,
+ 100).get();
+ }
+
+ /**
+ * Releases lock for the specific path. This operation is idempotent.
+ * @param exclusionPath exclusion path
+ */
+ protected void unlock(String exclusionPath) {
+ Tools.retryable(() -> lockMap.asJavaMap()
+ .compute(exclusionPath, (k, v) -> MutexState.evict(v, localNodeId)),
+ ConsistentMapException.ConcurrentModification.class,
+ Integer.MAX_VALUE,
+ 100).get();
+ }
+
+ /**
+ * Detects and releases all locks held by this node.
+ */
+ private void releaseOldLocks() {
+ Maps.filterValues(lockMap.asJavaMap(), state -> localNodeId.equals(state.holder()))
+ .keySet()
+ .forEach(path -> {
+ log.info("Detected zombie task still holding lock for {}. Releasing lock.", path);
+ unlock(path);
+ });
+ }
+
+ private class InternalLockMapEventListener implements MapEventListener<String, MutexState> {
+
+ @Override
+ public void event(MapEvent<String, MutexState> event) {
+ log.debug("Received {}", event);
+ if (event.type() == MapEvent.Type.UPDATE || event.type() == MapEvent.Type.INSERT) {
+ pending.computeIfPresent(event.key(), (k, future) -> {
+ MutexState state = Versioned.valueOrElse(event.value(), null);
+ if (state != null && localNodeId.equals(state.holder())) {
+ log.debug("Local node is now owner for {}", event.key());
+ future.complete(state);
+ return null;
+ } else {
+ return future;
+ }
+ });
+ InnerMutexTask task = activeTasks.get(event.key());
+ if (task != null && task.term() < Versioned.valueOrElse(event.value(), null).term()) {
+ task.stop();
+ }
+ }
+ }
+ }
+
+ private class InternalClusterEventListener implements ClusterEventListener {
+
+ @Override
+ public void event(ClusterEvent event) {
+ if (event.type() == ClusterEvent.Type.INSTANCE_DEACTIVATED ||
+ event.type() == ClusterEvent.Type.INSTANCE_REMOVED) {
+ NodeId nodeId = event.subject().id();
+ log.debug("{} is no longer active. Attemping to clean up its locks.", nodeId);
+ lockMap.asJavaMap().forEach((k, v) -> {
+ if (v.contains(nodeId)) {
+ lockMap.compute(k, (path, state) -> MutexState.evict(v, nodeId));
+ }
+ });
+ }
+ long activeNodes = clusterService.getNodes()
+ .stream()
+ .map(node -> clusterService.getState(node.id()))
+ .filter(State.ACTIVE::equals)
+ .count();
+ if (clusterService.getNodes().size() > 1 && activeNodes == 1) {
+ log.info("This node is partitioned away from the cluster. Stopping all inflight executions");
+ activeTasks.forEach((k, v) -> {
+ v.stop();
+ });
+ }
+ }
+ }
+
+ private static final class MutexState {
+
+ private final NodeId holder;
+ private final List<NodeId> waitList;
+ private final long term;
+
+ public static MutexState admit(MutexState state, NodeId nodeId) {
+ if (state == null) {
+ return new MutexState(nodeId, 1L, Lists.newArrayList());
+ } else if (state.holder() == null) {
+ return new MutexState(nodeId, state.term() + 1, Lists.newArrayList());
+ } else {
+ if (!state.contains(nodeId)) {
+ NodeId newHolder = state.holder();
+ List<NodeId> newWaitList = Lists.newArrayList(state.waitList());
+ newWaitList.add(nodeId);
+ return new MutexState(newHolder, state.term(), newWaitList);
+ } else {
+ return state;
+ }
+ }
+ }
+
+ public static MutexState evict(MutexState state, NodeId nodeId) {
+ return state.evict(nodeId);
+ }
+
+ public MutexState evict(NodeId nodeId) {
+ if (nodeId.equals(holder)) {
+ if (waitList.isEmpty()) {
+ return new MutexState(null, term, waitList);
+ }
+ List<NodeId> newWaitList = Lists.newArrayList(waitList);
+ NodeId newHolder = newWaitList.remove(0);
+ return new MutexState(newHolder, term + 1, newWaitList);
+ } else {
+ NodeId newHolder = holder;
+ List<NodeId> newWaitList = Lists.newArrayList(waitList);
+ newWaitList.remove(nodeId);
+ return new MutexState(newHolder, term, newWaitList);
+ }
+ }
+
+ public NodeId holder() {
+ return holder;
+ }
+
+ public List<NodeId> waitList() {
+ return waitList;
+ }
+
+ public long term() {
+ return term;
+ }
+
+ private boolean contains(NodeId nodeId) {
+ return (nodeId.equals(holder) || waitList.contains(nodeId));
+ }
+
+ private MutexState(NodeId holder, long term, List<NodeId> waitList) {
+ this.holder = holder;
+ this.term = term;
+ this.waitList = Lists.newArrayList(waitList);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("holder", holder)
+ .add("term", term)
+ .add("waitList", waitList)
+ .toString();
+ }
+ }
+
+ private class InnerMutexTask implements MutexTask {
+ private final MutexTask task;
+ private final String mutexPath;
+ private final long term;
+
+ public InnerMutexTask(String mutexPath, MutexTask task, long term) {
+ this.mutexPath = mutexPath;
+ this.term = term;
+ this.task = task;
+ }
+
+ public long term() {
+ return term;
+ }
+
+ @Override
+ public void start() {
+ log.debug("Starting execution for mutex task guarded by {}", mutexPath);
+ task.start();
+ log.debug("Finished execution for mutex task guarded by {}", mutexPath);
+ }
+
+ @Override
+ public void stop() {
+ log.debug("Stopping execution for mutex task guarded by {}", mutexPath);
+ task.stop();
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedDatabase.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedDatabase.java
new file mode 100644
index 0000000..fdcc59f
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PartitionedDatabase.java
@@ -0,0 +1,402 @@
+/*
+ * 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.service.DatabaseUpdate;
+import org.onosproject.store.service.Transaction;
+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);
+ }
+ }
+
+ @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<DatabaseUpdate>> perPartitionUpdates = Maps.newHashMap();
+ for (DatabaseUpdate update : transaction.updates()) {
+ Database partition = partitioner.getPartition(update.mapName(), update.key());
+ List<DatabaseUpdate> partitionUpdates =
+ perPartitionUpdates.computeIfAbsent(partition, k -> Lists.newLinkedList());
+ partitionUpdates.add(update);
+ }
+ Map<Database, Transaction> subTransactions = Maps.newHashMap();
+ perPartitionUpdates.forEach((k, v) -> subTransactions.put(k, new DefaultTransaction(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));
+ }
+}
+
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Partitioner.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Partitioner.java
new file mode 100644
index 0000000..7cabb3d
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Partitioner.java
@@ -0,0 +1,33 @@
+/*
+ * 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);
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PersistentStore.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PersistentStore.java
new file mode 100644
index 0000000..20381bd
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/PersistentStore.java
@@ -0,0 +1,47 @@
+/*
+ * 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.Map;
+
+/**
+ * A persistent store for an eventually consistent map.
+ */
+interface PersistentStore<K, V> {
+
+ /**
+ * Read the contents of the disk into the given maps.
+ *
+ * @param items items map
+ */
+ void readInto(Map<K, MapValue<V>> items);
+
+ /**
+ * Updates a key,value pair in the persistent store.
+ *
+ * @param key the key
+ * @param value the value
+ */
+ void update(K key, MapValue<V> value);
+
+ /**
+ * Removes a key from persistent store.
+ *
+ * @param key the key to remove
+ */
+ void remove(K key);
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Result.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Result.java
new file mode 100644
index 0000000..acd724c
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/Result.java
@@ -0,0 +1,121 @@
+/*
+ * 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();
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleKeyHashPartitioner.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleKeyHashPartitioner.java
new file mode 100644
index 0000000..c94090a
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleKeyHashPartitioner.java
@@ -0,0 +1,38 @@
+/*
+ * 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());
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleTableHashPartitioner.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleTableHashPartitioner.java
new file mode 100644
index 0000000..bf7971e
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/SimpleTableHashPartitioner.java
@@ -0,0 +1,39 @@
+/*
+ * 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());
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StateMachineUpdate.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StateMachineUpdate.java
new file mode 100644
index 0000000..6d1dcd7
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StateMachineUpdate.java
@@ -0,0 +1,91 @@
+/*
+ * 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();
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java
new file mode 100644
index 0000000..7cfc88b
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/TransactionManager.java
@@ -0,0 +1,126 @@
+/*
+ * 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.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import org.onlab.util.KryoNamespace;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncConsistentMap;
+import org.onosproject.store.service.ConsistentMapBuilder;
+import org.onosproject.store.service.DatabaseUpdate;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.Transaction;
+import org.onosproject.store.service.Versioned;
+import org.onosproject.store.service.Transaction.State;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Agent that runs the two phase commit protocol.
+ */
+public class TransactionManager {
+
+ private static final KryoNamespace KRYO_NAMESPACE = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.BASIC)
+ .nextId(KryoNamespace.FLOATING_ID)
+ .register(Versioned.class)
+ .register(DatabaseUpdate.class)
+ .register(DatabaseUpdate.Type.class)
+ .register(DefaultTransaction.class)
+ .register(Transaction.State.class)
+ .build();
+
+ private final Serializer serializer = Serializer.using(Arrays.asList(KRYO_NAMESPACE));
+ private final Database database;
+ private final AsyncConsistentMap<Long, Transaction> transactions;
+
+ /**
+ * Constructs a new TransactionManager for the specified database instance.
+ *
+ * @param database database
+ * @param mapBuilder builder for ConsistentMap instances
+ */
+ public TransactionManager(Database database, ConsistentMapBuilder<Long, Transaction> mapBuilder) {
+ this.database = checkNotNull(database, "database cannot be null");
+ this.transactions = mapBuilder.withName("onos-transactions")
+ .withSerializer(serializer)
+ .buildAsyncMap();
+ }
+
+ /**
+ * Executes the specified transaction by employing a two phase commit protocol.
+ *
+ * @param transaction transaction to commit
+ * @return transaction result. Result value true indicates a successful commit, false
+ * indicates abort
+ */
+ public CompletableFuture<CommitResponse> execute(Transaction transaction) {
+ // clean up if this transaction in already in a terminal state.
+ if (transaction.state() == Transaction.State.COMMITTED ||
+ transaction.state() == Transaction.State.ROLLEDBACK) {
+ return transactions.remove(transaction.id()).thenApply(v -> CommitResponse.success(ImmutableList.of()));
+ } else if (transaction.state() == Transaction.State.COMMITTING) {
+ return commit(transaction);
+ } else if (transaction.state() == Transaction.State.ROLLINGBACK) {
+ return rollback(transaction).thenApply(v -> CommitResponse.success(ImmutableList.of()));
+ } else {
+ return prepare(transaction).thenCompose(v -> v ? commit(transaction) : rollback(transaction));
+ }
+ }
+
+
+ /**
+ * Returns all transactions in the system.
+ *
+ * @return future for a collection of transactions
+ */
+ public CompletableFuture<Collection<Transaction>> getTransactions() {
+ return transactions.values().thenApply(c -> {
+ Collection<Transaction> txns = c.stream().map(v -> v.value()).collect(Collectors.toList());
+ return txns;
+ });
+ }
+
+ 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 ? State.COMMITTING : State.ROLLINGBACK))
+ .thenApply(v -> status));
+ }
+
+ private CompletableFuture<CommitResponse> commit(Transaction transaction) {
+ return database.commit(transaction)
+ .whenComplete((r, e) -> transactions.put(
+ transaction.id(),
+ transaction.transition(Transaction.State.COMMITTED)));
+ }
+
+ private CompletableFuture<CommitResponse> rollback(Transaction transaction) {
+ return database.rollback(transaction)
+ .thenCompose(v -> transactions.put(
+ transaction.id(),
+ transaction.transition(Transaction.State.ROLLEDBACK)))
+ .thenApply(v -> CommitResponse.failure());
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateEntry.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateEntry.java
new file mode 100644
index 0000000..eda2ba4
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateEntry.java
@@ -0,0 +1,80 @@
+/*
+ * 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 com.google.common.base.MoreObjects;
+
+/**
+ * Describes a single update event in an EventuallyConsistentMap.
+ */
+final class UpdateEntry<K, V> {
+ private final K key;
+ private final MapValue<V> value;
+
+ /**
+ * Creates a new update entry.
+ *
+ * @param key key of the entry
+ * @param value value of the entry
+ */
+ public UpdateEntry(K key, MapValue<V> value) {
+ this.key = checkNotNull(key);
+ this.value = value;
+ }
+
+ /**
+ * Returns the key.
+ *
+ * @return the key
+ */
+ public K key() {
+ return key;
+ }
+
+ /**
+ * Returns the value of the entry.
+ *
+ * @return the value
+ */
+ public MapValue<V> value() {
+ return value;
+ }
+
+ /**
+ * Returns if this entry is newer than other entry.
+ * @param other other entry
+ * @return true if this entry is newer; false otherwise
+ */
+ public boolean isNewerThan(UpdateEntry<K, V> other) {
+ return other == null || other.value == null || (value != null && value.isNewerThan(other.value));
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(getClass())
+ .add("key", key())
+ .add("value", value)
+ .toString();
+ }
+
+ @SuppressWarnings("unused")
+ private UpdateEntry() {
+ this.key = null;
+ this.value = null;
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateResult.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateResult.java
new file mode 100644
index 0000000..cd943ba
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/UpdateResult.java
@@ -0,0 +1,85 @@
+/*
+ * 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 {
+ MapEvent.Type eventType = oldValue == null ?
+ MapEvent.Type.INSERT : newValue == null ? MapEvent.Type.REMOVE : MapEvent.Type.UPDATE;
+ Versioned<V> eventValue = eventType == MapEvent.Type.REMOVE ? oldValue : newValue;
+ return new MapEvent<>(mapName(), eventType, key(), eventValue);
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/package-info.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/package-info.java
index 0176cb5..cb20c3e 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/package-info.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/package-info.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2016 Open Networking Laboratory
+ * 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.
@@ -15,6 +15,7 @@
*/
/**
- * Implementation for distributed state management primitives.
+ * Implementation of partitioned and distributed store facility capable of
+ * providing consistent update semantics.
*/
-package org.onosproject.store.primitives.impl;
+package org.onosproject.store.primitives.impl;
\ No newline at end of file