홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Madan Jampani
Committed by Gerrit Code Review
832686da 1 parent d4489882

Post DatabaseManager deprecation code cleanup 

- Dropping MutexExecutionService as there are now better alternatives
- Dropping New from class names that were added during transition phase

Change-Id: If0cdd3321081c3f8fda81441ef2c84549b616edd
Inline Side-by-side
Showing 16 changed files with 44 additions and 523 deletions
...@@ -21,9 +21,7 @@ import org.apache.karaf.shell.commands.Command; ...@@ -21,9 +21,7 @@ import org.apache.karaf.shell.commands.Command;
21 import org.onosproject.cli.AbstractShellCommand; 21 import org.onosproject.cli.AbstractShellCommand;
22 import org.onosproject.store.service.StorageAdminService; 22 import org.onosproject.store.service.StorageAdminService;
23 23
24 -import com.fasterxml.jackson.databind.JsonNode;
25 import com.fasterxml.jackson.databind.ObjectMapper; 24 import com.fasterxml.jackson.databind.ObjectMapper;
26 -import com.fasterxml.jackson.databind.node.ArrayNode;
27 import com.fasterxml.jackson.databind.node.ObjectNode; 25 import com.fasterxml.jackson.databind.node.ObjectNode;
28 26
29 /** 27 /**
...@@ -35,73 +33,17 @@ public class CountersListCommand extends AbstractShellCommand { ...@@ -35,73 +33,17 @@ public class CountersListCommand extends AbstractShellCommand {
35 33
36 private static final String FMT = "name=%s value=%d"; 34 private static final String FMT = "name=%s value=%d";
37 35
38 - /**
39 - * Displays counters as text.
40 - *
41 - * @param counters counter info
42 - */
43 - private void displayCounters(Map<String, Long> counters) {
44 - counters.forEach((name, value) -> print(FMT, name, value));
45 - }
46 -
47 - /**
48 - * Converts info for counters into a JSON object.
49 - *
50 - * @param counters counter info
51 - */
52 - private JsonNode json(Map<String, Long> counters) {
53 - ObjectMapper mapper = new ObjectMapper();
54 - ArrayNode jsonCounters = mapper.createArrayNode();
55 -
56 - // Create a JSON node for each counter
57 - counters.forEach((name, value) -> {
58 - ObjectNode jsonCounter = mapper.createObjectNode();
59 - jsonCounter.put("name", name)
60 - .put("value", value);
61 - jsonCounters.add(jsonCounter);
62 - });
63 -
64 - return jsonCounters;
65 - }
66 -
67 - /**
68 - * Converts info for counters from different databases into a JSON object.
69 - *
70 - * @param partitionedDbCounters counters info
71 - * @param inMemoryDbCounters counters info
72 - */
73 - private JsonNode jsonAllCounters(Map<String, Long> partitionedDbCounters,
74 - Map<String, Long> inMemoryDbCounters) {
75 - ObjectMapper mapper = new ObjectMapper();
76 - ArrayNode jsonCounters = mapper.createArrayNode();
77 -
78 - // Create a JSON node for partitioned database counter
79 - ObjectNode jsonPartitionedDatabaseCounters = mapper.createObjectNode();
80 - jsonPartitionedDatabaseCounters.set("partitionedDatabaseCounters",
81 - json(partitionedDbCounters));
82 - jsonCounters.add(jsonPartitionedDatabaseCounters);
83 - // Create a JSON node for in-memory database counter
84 - ObjectNode jsonInMemoryDatabseCounters = mapper.createObjectNode();
85 - jsonInMemoryDatabseCounters.set("inMemoryDatabaseCounters",
86 - json(inMemoryDbCounters));
87 - jsonCounters.add(jsonInMemoryDatabseCounters);
88 -
89 - return jsonCounters;
90 - }
91 -
92 -
93 @Override 36 @Override
94 protected void execute() { 37 protected void execute() {
95 StorageAdminService storageAdminService = get(StorageAdminService.class); 38 StorageAdminService storageAdminService = get(StorageAdminService.class);
96 - Map<String, Long> partitionedDatabaseCounters = storageAdminService.getPartitionedDatabaseCounters(); 39 + Map<String, Long> counters = storageAdminService.getCounters();
97 - Map<String, Long> inMemoryDatabaseCounters = storageAdminService.getInMemoryDatabaseCounters();
98 if (outputJson()) { 40 if (outputJson()) {
99 - print("%s", jsonAllCounters(partitionedDatabaseCounters, inMemoryDatabaseCounters)); 41 + ObjectMapper mapper = new ObjectMapper();
42 + ObjectNode jsonCounters = mapper.createObjectNode();
43 + counters.forEach((k, v) -> jsonCounters.put(k, v));
44 + print("%s", jsonCounters);
100 } else { 45 } else {
101 - print("Partitioned database counters:"); 46 + counters.keySet().stream().sorted().forEach(name -> print(FMT, name, counters.get(name)));
102 - displayCounters(partitionedDatabaseCounters);
103 - print("In-memory database counters:");
104 - displayCounters(inMemoryDatabaseCounters);
105 } 47 }
106 } 48 }
107 } 49 }
......
...@@ -51,24 +51,6 @@ public interface DistributedQueueBuilder<E> { ...@@ -51,24 +51,6 @@ public interface DistributedQueueBuilder<E> {
51 DistributedQueueBuilder<E> withSerializer(Serializer serializer); 51 DistributedQueueBuilder<E> withSerializer(Serializer serializer);
52 52
53 /** 53 /**
54 - *
55 - *
56 - * @return this DistributedQueueBuilder for method chaining
57 - */
58 - DistributedQueueBuilder<E> withMeteringDisabled();
59 -
60 -
61 - /**
62 - * Disables persistence of queues entries.
63 - * <p>
64 - * When persistence is disabled, a full cluster restart will wipe out all
65 - * queue entries.
66 - * </p>
67 - * @return this DistributedQueueBuilder for method chaining
68 - */
69 - DistributedQueueBuilder<E> withPersistenceDisabled();
70 -
71 - /**
72 * Builds a queue based on the configuration options 54 * Builds a queue based on the configuration options
73 * supplied to this builder. 55 * supplied to this builder.
74 * 56 *
......
1 -/*
2 - * Copyright 2015 Open Networking Laboratory
3 - *
4 - * Licensed under the Apache License, Version 2.0 (the "License");
5 - * you may not use this file except in compliance with the License.
6 - * You may obtain a copy of the License at
7 - *
8 - * http://www.apache.org/licenses/LICENSE-2.0
9 - *
10 - * Unless required by applicable law or agreed to in writing, software
11 - * distributed under the License is distributed on an "AS IS" BASIS,
12 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 - * See the License for the specific language governing permissions and
14 - * limitations under the License.
15 - */
16 -package org.onosproject.store.service;
17 -
18 -import java.util.concurrent.CompletableFuture;
19 -import java.util.concurrent.Executor;
20 -
21 -/**
22 - * Service for mutually exclusive job execution.
23 - */
24 -public interface MutexExecutionService {
25 -
26 - /**
27 - * Runs the specified task in a mutually exclusive fashion.
28 - * @param task task to run
29 - * @param exclusionPath path on which different instances synchronize
30 - * @param executor executor to use for running the task
31 - * @return future that is completed when the task execution completes.
32 - */
33 - CompletableFuture<Void> execute(MutexTask task, String exclusionPath, Executor executor);
34 -}
...\ No newline at end of file ...\ No newline at end of file
...@@ -46,34 +46,12 @@ public interface StorageAdminService { ...@@ -46,34 +46,12 @@ public interface StorageAdminService {
46 46
47 /** 47 /**
48 * Returns information about all the atomic counters in the system. 48 * Returns information about all the atomic counters in the system.
49 - * If 2 counters belonging to 2 different databases have the same name,
50 - * then only one counter from one database is returned.
51 * 49 *
52 * @return mapping from counter name to that counter's next value 50 * @return mapping from counter name to that counter's next value
53 - * @deprecated 1.5.0 Falcon Release
54 */ 51 */
55 - @Deprecated
56 Map<String, Long> getCounters(); 52 Map<String, Long> getCounters();
57 53
58 /** 54 /**
59 - * Returns information about all the atomic partitioned database counters in the system.
60 - *
61 - * @return mapping from counter name to that counter's next value
62 - * @deprecated 1.5.0 Falcon Release
63 - */
64 - @Deprecated
65 - Map<String, Long> getPartitionedDatabaseCounters();
66 -
67 - /**
68 - * Returns information about all the atomic in-memory database counters in the system.
69 - *
70 - * @return mapping from counter name to that counter's next value
71 - * @deprecated 1.5.0 Falcon Release
72 - */
73 - @Deprecated
74 - Map<String, Long> getInMemoryDatabaseCounters();
75 -
76 - /**
77 * Returns all pending transactions. 55 * Returns all pending transactions.
78 * 56 *
79 * @return collection of pending transaction identifiers. 57 * @return collection of pending transaction identifiers.
......
...@@ -73,7 +73,6 @@ import org.onosproject.store.cluster.messaging.MessagingService; ...@@ -73,7 +73,6 @@ import org.onosproject.store.cluster.messaging.MessagingService;
73 import org.onosproject.store.primitives.PartitionAdminService; 73 import org.onosproject.store.primitives.PartitionAdminService;
74 import org.onosproject.store.primitives.PartitionService; 74 import org.onosproject.store.primitives.PartitionService;
75 import org.onosproject.store.service.LogicalClockService; 75 import org.onosproject.store.service.LogicalClockService;
76 -import org.onosproject.store.service.MutexExecutionService;
77 import org.onosproject.store.service.StorageAdminService; 76 import org.onosproject.store.service.StorageAdminService;
78 import org.onosproject.store.service.StorageService; 77 import org.onosproject.store.service.StorageService;
79 import org.onosproject.ui.UiExtensionService; 78 import org.onosproject.ui.UiExtensionService;
...@@ -248,7 +247,6 @@ public final class DefaultPolicyBuilder { ...@@ -248,7 +247,6 @@ public final class DefaultPolicyBuilder {
248 permSet.add(new ServicePermission(MessagingService.class.getName(), ServicePermission.GET)); 247 permSet.add(new ServicePermission(MessagingService.class.getName(), ServicePermission.GET));
249 permSet.add(new ServicePermission(PartitionService.class.getName(), ServicePermission.GET)); 248 permSet.add(new ServicePermission(PartitionService.class.getName(), ServicePermission.GET));
250 permSet.add(new ServicePermission(LogicalClockService.class.getName(), ServicePermission.GET)); 249 permSet.add(new ServicePermission(LogicalClockService.class.getName(), ServicePermission.GET));
251 - permSet.add(new ServicePermission(MutexExecutionService.class.getName(), ServicePermission.GET));
252 permSet.add(new ServicePermission(StorageService.class.getName(), ServicePermission.GET)); 250 permSet.add(new ServicePermission(StorageService.class.getName(), ServicePermission.GET));
253 permSet.add(new ServicePermission(UiExtensionService.class.getName(), ServicePermission.GET)); 251 permSet.add(new ServicePermission(UiExtensionService.class.getName(), ServicePermission.GET));
254 252
...@@ -376,8 +374,6 @@ public final class DefaultPolicyBuilder { ...@@ -376,8 +374,6 @@ public final class DefaultPolicyBuilder {
376 PartitionService.class.getName())); 374 PartitionService.class.getName()));
377 serviceDirectory.put(CLOCK_WRITE, ImmutableSet.of( 375 serviceDirectory.put(CLOCK_WRITE, ImmutableSet.of(
378 LogicalClockService.class.getName())); 376 LogicalClockService.class.getName()));
379 - serviceDirectory.put(MUTEX_WRITE, ImmutableSet.of(
380 - MutexExecutionService.class.getName()));
381 377
382 return serviceDirectory; 378 return serviceDirectory;
383 } 379 }
......
...@@ -34,7 +34,11 @@ import com.google.common.base.Throwables; ...@@ -34,7 +34,11 @@ import com.google.common.base.Throwables;
34 import com.google.common.collect.Maps; 34 import com.google.common.collect.Maps;
35 import org.onosproject.utils.MeteringAgent; 35 import org.onosproject.utils.MeteringAgent;
36 36
37 - 37 +/**
38 + * Default implementation of a {@code AsyncAtomicValue}.
39 + *
40 + * @param <V> value type
41 + */
38 public class DefaultAsyncAtomicValue<V> implements AsyncAtomicValue<V> { 42 public class DefaultAsyncAtomicValue<V> implements AsyncAtomicValue<V> {
39 43
40 private final String name; 44 private final String name;
......
...@@ -22,11 +22,11 @@ import org.onosproject.store.service.AtomicCounterBuilder; ...@@ -22,11 +22,11 @@ import org.onosproject.store.service.AtomicCounterBuilder;
22 /** 22 /**
23 * Default implementation of AtomicCounterBuilder. 23 * Default implementation of AtomicCounterBuilder.
24 */ 24 */
25 -public class NewDefaultAtomicCounterBuilder extends AtomicCounterBuilder { 25 +public class DefaultAtomicCounterBuilder extends AtomicCounterBuilder {
26 26
27 private final DistributedPrimitiveCreator primitiveCreator; 27 private final DistributedPrimitiveCreator primitiveCreator;
28 28
29 - public NewDefaultAtomicCounterBuilder(DistributedPrimitiveCreator primitiveCreator) { 29 + public DefaultAtomicCounterBuilder(DistributedPrimitiveCreator primitiveCreator) {
30 this.primitiveCreator = primitiveCreator; 30 this.primitiveCreator = primitiveCreator;
31 } 31 }
32 32
......
...@@ -26,11 +26,11 @@ import org.onosproject.store.service.ConsistentMapBuilder; ...@@ -26,11 +26,11 @@ import org.onosproject.store.service.ConsistentMapBuilder;
26 * @param <K> type for map key 26 * @param <K> type for map key
27 * @param <V> type for map value 27 * @param <V> type for map value
28 */ 28 */
29 -public class NewDefaultConsistentMapBuilder<K, V> extends ConsistentMapBuilder<K, V> { 29 +public class DefaultConsistentMapBuilder<K, V> extends ConsistentMapBuilder<K, V> {
30 30
31 private final DistributedPrimitiveCreator primitiveCreator; 31 private final DistributedPrimitiveCreator primitiveCreator;
32 32
33 - public NewDefaultConsistentMapBuilder(DistributedPrimitiveCreator primitiveCreator) { 33 + public DefaultConsistentMapBuilder(DistributedPrimitiveCreator primitiveCreator) {
34 this.primitiveCreator = primitiveCreator; 34 this.primitiveCreator = primitiveCreator;
35 } 35 }
36 36
......
...@@ -28,15 +28,13 @@ import static com.google.common.base.Preconditions.checkState; ...@@ -28,15 +28,13 @@ import static com.google.common.base.Preconditions.checkState;
28 * 28 *
29 * @param <E> queue entry type 29 * @param <E> queue entry type
30 */ 30 */
31 -public class NewDefaultDistributedQueueBuilder<E> implements DistributedQueueBuilder<E> { 31 +public class DefaultDistributedQueueBuilder<E> implements DistributedQueueBuilder<E> {
32 32
33 private final DistributedPrimitiveCreator primitiveCreator; 33 private final DistributedPrimitiveCreator primitiveCreator;
34 private String name; 34 private String name;
35 - private boolean persistenceEnabled = true;
36 - private boolean metering = true;
37 private Serializer serializer; 35 private Serializer serializer;
38 36
39 - public NewDefaultDistributedQueueBuilder(DistributedPrimitiveCreator primitiveCreator) { 37 + public DefaultDistributedQueueBuilder(DistributedPrimitiveCreator primitiveCreator) {
40 this.primitiveCreator = primitiveCreator; 38 this.primitiveCreator = primitiveCreator;
41 } 39 }
42 40
...@@ -54,18 +52,6 @@ public class NewDefaultDistributedQueueBuilder<E> implements DistributedQueueBui ...@@ -54,18 +52,6 @@ public class NewDefaultDistributedQueueBuilder<E> implements DistributedQueueBui
54 return this; 52 return this;
55 } 53 }
56 54
57 - @Override
58 - public DistributedQueueBuilder<E> withMeteringDisabled() {
59 - metering = false;
60 - return this;
61 - }
62 -
63 - @Override
64 - public DistributedQueueBuilder<E> withPersistenceDisabled() {
65 - persistenceEnabled = false;
66 - return this;
67 - }
68 -
69 private boolean validInputs() { 55 private boolean validInputs() {
70 return name != null && serializer != null; 56 return name != null && serializer != null;
71 } 57 }
......
...@@ -32,7 +32,7 @@ import com.google.common.collect.Sets; ...@@ -32,7 +32,7 @@ import com.google.common.collect.Sets;
32 /** 32 /**
33 * Default implementation of transaction context. 33 * Default implementation of transaction context.
34 */ 34 */
35 -public class NewDefaultTransactionContext implements TransactionContext { 35 +public class DefaultTransactionContext implements TransactionContext {
36 36
37 private final AtomicBoolean isOpen = new AtomicBoolean(false); 37 private final AtomicBoolean isOpen = new AtomicBoolean(false);
38 private final DistributedPrimitiveCreator creator; 38 private final DistributedPrimitiveCreator creator;
...@@ -41,7 +41,7 @@ public class NewDefaultTransactionContext implements TransactionContext { ...@@ -41,7 +41,7 @@ public class NewDefaultTransactionContext implements TransactionContext {
41 private final Set<TransactionParticipant> txParticipants = Sets.newConcurrentHashSet(); 41 private final Set<TransactionParticipant> txParticipants = Sets.newConcurrentHashSet();
42 private final MeteringAgent monitor; 42 private final MeteringAgent monitor;
43 43
44 - public NewDefaultTransactionContext(TransactionId transactionId, 44 + public DefaultTransactionContext(TransactionId transactionId,
45 DistributedPrimitiveCreator creator, 45 DistributedPrimitiveCreator creator,
46 TransactionCoordinator transactionCoordinator) { 46 TransactionCoordinator transactionCoordinator) {
47 this.transactionId = transactionId; 47 this.transactionId = transactionId;
......
...@@ -22,13 +22,13 @@ import org.onosproject.store.service.TransactionContextBuilder; ...@@ -22,13 +22,13 @@ import org.onosproject.store.service.TransactionContextBuilder;
22 /** 22 /**
23 * Default Transaction Context Builder. 23 * Default Transaction Context Builder.
24 */ 24 */
25 -public class NewDefaultTransactionContextBuilder extends TransactionContextBuilder { 25 +public class DefaultTransactionContextBuilder extends TransactionContextBuilder {
26 26
27 private final TransactionId transactionId; 27 private final TransactionId transactionId;
28 private final DistributedPrimitiveCreator primitiveCreator; 28 private final DistributedPrimitiveCreator primitiveCreator;
29 private final TransactionCoordinator transactionCoordinator; 29 private final TransactionCoordinator transactionCoordinator;
30 30
31 - public NewDefaultTransactionContextBuilder(TransactionId transactionId, 31 + public DefaultTransactionContextBuilder(TransactionId transactionId,
32 DistributedPrimitiveCreator primitiveCreator, 32 DistributedPrimitiveCreator primitiveCreator,
33 TransactionCoordinator transactionCoordinator) { 33 TransactionCoordinator transactionCoordinator) {
34 this.transactionId = transactionId; 34 this.transactionId = transactionId;
...@@ -38,7 +38,7 @@ public class NewDefaultTransactionContextBuilder extends TransactionContextBuild ...@@ -38,7 +38,7 @@ public class NewDefaultTransactionContextBuilder extends TransactionContextBuild
38 38
39 @Override 39 @Override
40 public TransactionContext build() { 40 public TransactionContext build() {
41 - return new NewDefaultTransactionContext(transactionId, 41 + return new DefaultTransactionContext(transactionId,
42 primitiveCreator, 42 primitiveCreator,
43 transactionCoordinator); 43 transactionCoordinator);
44 } 44 }
......
1 -/*
2 - * Copyright 2015 Open Networking Laboratory
3 - *
4 - * Licensed under the Apache License, Version 2.0 (the "License");
5 - * you may not use this file except in compliance with the License.
6 - * You may obtain a copy of the License at
7 - *
8 - * http://www.apache.org/licenses/LICENSE-2.0
9 - *
10 - * Unless required by applicable law or agreed to in writing, software
11 - * distributed under the License is distributed on an "AS IS" BASIS,
12 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 - * See the License for the specific language governing permissions and
14 - * limitations under the License.
15 - */
16 -package org.onosproject.store.primitives.impl;
17 -
18 -import static org.slf4j.LoggerFactory.getLogger;
19 -
20 -import java.util.Arrays;
21 -import java.util.List;
22 -import java.util.Map;
23 -import java.util.concurrent.CompletableFuture;
24 -import java.util.concurrent.Executor;
25 -
26 -import org.apache.felix.scr.annotations.Activate;
27 -import org.apache.felix.scr.annotations.Component;
28 -import org.apache.felix.scr.annotations.Deactivate;
29 -import org.apache.felix.scr.annotations.Reference;
30 -import org.apache.felix.scr.annotations.ReferenceCardinality;
31 -import org.apache.felix.scr.annotations.Service;
32 -import org.onlab.util.Tools;
33 -import org.onosproject.cluster.ClusterEvent;
34 -import org.onosproject.cluster.ClusterEventListener;
35 -import org.onosproject.cluster.ClusterService;
36 -import org.onosproject.cluster.ControllerNode.State;
37 -import org.onosproject.cluster.NodeId;
38 -import org.onosproject.store.serializers.KryoNamespaces;
39 -import org.onosproject.store.service.ConsistentMap;
40 -import org.onosproject.store.service.ConsistentMapException;
41 -import org.onosproject.store.service.MapEvent;
42 -import org.onosproject.store.service.MapEventListener;
43 -import org.onosproject.store.service.MutexExecutionService;
44 -import org.onosproject.store.service.MutexTask;
45 -import org.onosproject.store.service.Serializer;
46 -import org.onosproject.store.service.StorageService;
47 -import org.onosproject.store.service.Versioned;
48 -import org.slf4j.Logger;
49 -
50 -import com.google.common.base.MoreObjects;
51 -import com.google.common.collect.Lists;
52 -import com.google.common.collect.Maps;
53 -import static org.onosproject.security.AppGuard.checkPermission;
54 -import static org.onosproject.security.AppPermission.Type.MUTEX_WRITE;
55 -/**
56 - * Implementation of a MutexExecutionService.
57 - */
58 -@Component(immediate = true)
59 -@Service
60 -public class MutexExecutionManager implements MutexExecutionService {
61 -
62 - private final Logger log = getLogger(getClass());
63 -
64 - protected ConsistentMap<String, MutexState> lockMap;
65 - protected NodeId localNodeId;
66 -
67 - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
68 - protected ClusterService clusterService;
69 -
70 - @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
71 - protected StorageService storageService;
72 -
73 - private final MapEventListener<String, MutexState> mapEventListener = new InternalLockMapEventListener();
74 - private final ClusterEventListener clusterEventListener = new InternalClusterEventListener();
75 -
76 - private Map<String, CompletableFuture<MutexState>> pending = Maps.newConcurrentMap();
77 - private Map<String, InnerMutexTask> activeTasks = Maps.newConcurrentMap();
78 -
79 - @Activate
80 - public void activate() {
81 - localNodeId = clusterService.getLocalNode().id();
82 - lockMap = storageService.<String, MutexState>consistentMapBuilder()
83 - .withName("onos-mutexes")
84 - .withSerializer(Serializer.using(Arrays.asList(KryoNamespaces.API), MutexState.class))
85 - .withPartitionsDisabled()
86 - .build();
87 - lockMap.addListener(mapEventListener);
88 - clusterService.addListener(clusterEventListener);
89 - releaseOldLocks();
90 - log.info("Started");
91 - }
92 -
93 - @Deactivate
94 - public void deactivate() {
95 - lockMap.removeListener(mapEventListener);
96 - pending.values().forEach(future -> future.cancel(true));
97 - activeTasks.forEach((k, v) -> {
98 - v.stop();
99 - unlock(k);
100 - });
101 - clusterService.removeListener(clusterEventListener);
102 - log.info("Stopped");
103 - }
104 -
105 - @Override
106 - public CompletableFuture<Void> execute(MutexTask task, String exclusionPath, Executor executor) {
107 - checkPermission(MUTEX_WRITE);
108 - return lock(exclusionPath)
109 - .thenApply(state -> activeTasks.computeIfAbsent(exclusionPath,
110 - k -> new InnerMutexTask(exclusionPath,
111 - task,
112 - state.term())))
113 - .thenAcceptAsync(t -> t.start(), executor)
114 - .whenComplete((r, e) -> unlock(exclusionPath));
115 - }
116 -
117 - protected CompletableFuture<MutexState> lock(String exclusionPath) {
118 - CompletableFuture<MutexState> future =
119 - pending.computeIfAbsent(exclusionPath, k -> new CompletableFuture<>());
120 - tryLock(exclusionPath);
121 - return future;
122 - }
123 -
124 - /**
125 - * Attempts to acquire lock for a path. If lock is held by some other node, adds this node to
126 - * the wait list.
127 - * @param exclusionPath exclusion path
128 - */
129 - protected void tryLock(String exclusionPath) {
130 - Tools.retryable(() -> lockMap.asJavaMap()
131 - .compute(exclusionPath,
132 - (k, v) -> MutexState.admit(v, localNodeId)),
133 - ConsistentMapException.ConcurrentModification.class,
134 - Integer.MAX_VALUE,
135 - 100).get();
136 - }
137 -
138 - /**
139 - * Releases lock for the specific path. This operation is idempotent.
140 - * @param exclusionPath exclusion path
141 - */
142 - protected void unlock(String exclusionPath) {
143 - Tools.retryable(() -> lockMap.asJavaMap()
144 - .compute(exclusionPath, (k, v) -> MutexState.evict(v, localNodeId)),
145 - ConsistentMapException.ConcurrentModification.class,
146 - Integer.MAX_VALUE,
147 - 100).get();
148 - }
149 -
150 - /**
151 - * Detects and releases all locks held by this node.
152 - */
153 - private void releaseOldLocks() {
154 - Maps.filterValues(lockMap.asJavaMap(), state -> localNodeId.equals(state.holder()))
155 - .keySet()
156 - .forEach(path -> {
157 - log.info("Detected zombie task still holding lock for {}. Releasing lock.", path);
158 - unlock(path);
159 - });
160 - }
161 -
162 - private class InternalLockMapEventListener implements MapEventListener<String, MutexState> {
163 -
164 - @Override
165 - public void event(MapEvent<String, MutexState> event) {
166 - log.debug("Received {}", event);
167 - if (event.type() == MapEvent.Type.UPDATE || event.type() == MapEvent.Type.INSERT) {
168 - pending.computeIfPresent(event.key(), (k, future) -> {
169 - MutexState state = Versioned.valueOrElse(event.value(), null);
170 - if (state != null && localNodeId.equals(state.holder())) {
171 - log.debug("Local node is now owner for {}", event.key());
172 - future.complete(state);
173 - return null;
174 - } else {
175 - return future;
176 - }
177 - });
178 - InnerMutexTask task = activeTasks.get(event.key());
179 - if (task != null && task.term() < Versioned.valueOrElse(event.value(), null).term()) {
180 - task.stop();
181 - }
182 - }
183 - }
184 - }
185 -
186 - private class InternalClusterEventListener implements ClusterEventListener {
187 -
188 - @Override
189 - public void event(ClusterEvent event) {
190 - if (event.type() == ClusterEvent.Type.INSTANCE_DEACTIVATED ||
191 - event.type() == ClusterEvent.Type.INSTANCE_REMOVED) {
192 - NodeId nodeId = event.subject().id();
193 - log.debug("{} is no longer active. Attemping to clean up its locks.", nodeId);
194 - lockMap.asJavaMap().forEach((k, v) -> {
195 - if (v.contains(nodeId)) {
196 - lockMap.compute(k, (path, state) -> MutexState.evict(v, nodeId));
197 - }
198 - });
199 - }
200 - long activeNodes = clusterService.getNodes()
201 - .stream()
202 - .map(node -> clusterService.getState(node.id()))
203 - .filter(State::isActive)
204 - .count();
205 - if (clusterService.getNodes().size() > 1 && activeNodes == 1) {
206 - log.info("This node is partitioned away from the cluster. Stopping all inflight executions");
207 - activeTasks.forEach((k, v) -> {
208 - v.stop();
209 - });
210 - }
211 - }
212 - }
213 -
214 - private static final class MutexState {
215 -
216 - private final NodeId holder;
217 - private final List<NodeId> waitList;
218 - private final long term;
219 -
220 - public static MutexState admit(MutexState state, NodeId nodeId) {
221 - if (state == null) {
222 - return new MutexState(nodeId, 1L, Lists.newArrayList());
223 - } else if (state.holder() == null) {
224 - return new MutexState(nodeId, state.term() + 1, Lists.newArrayList());
225 - } else {
226 - if (!state.contains(nodeId)) {
227 - NodeId newHolder = state.holder();
228 - List<NodeId> newWaitList = Lists.newArrayList(state.waitList());
229 - newWaitList.add(nodeId);
230 - return new MutexState(newHolder, state.term(), newWaitList);
231 - } else {
232 - return state;
233 - }
234 - }
235 - }
236 -
237 - public static MutexState evict(MutexState state, NodeId nodeId) {
238 - return state.evict(nodeId);
239 - }
240 -
241 - public MutexState evict(NodeId nodeId) {
242 - if (nodeId.equals(holder)) {
243 - if (waitList.isEmpty()) {
244 - return new MutexState(null, term, waitList);
245 - }
246 - List<NodeId> newWaitList = Lists.newArrayList(waitList);
247 - NodeId newHolder = newWaitList.remove(0);
248 - return new MutexState(newHolder, term + 1, newWaitList);
249 - } else {
250 - NodeId newHolder = holder;
251 - List<NodeId> newWaitList = Lists.newArrayList(waitList);
252 - newWaitList.remove(nodeId);
253 - return new MutexState(newHolder, term, newWaitList);
254 - }
255 - }
256 -
257 - public NodeId holder() {
258 - return holder;
259 - }
260 -
261 - public List<NodeId> waitList() {
262 - return waitList;
263 - }
264 -
265 - public long term() {
266 - return term;
267 - }
268 -
269 - private boolean contains(NodeId nodeId) {
270 - return (nodeId.equals(holder) || waitList.contains(nodeId));
271 - }
272 -
273 - private MutexState(NodeId holder, long term, List<NodeId> waitList) {
274 - this.holder = holder;
275 - this.term = term;
276 - this.waitList = Lists.newArrayList(waitList);
277 - }
278 -
279 - @Override
280 - public String toString() {
281 - return MoreObjects.toStringHelper(getClass())
282 - .add("holder", holder)
283 - .add("term", term)
284 - .add("waitList", waitList)
285 - .toString();
286 - }
287 - }
288 -
289 - private class InnerMutexTask implements MutexTask {
290 - private final MutexTask task;
291 - private final String mutexPath;
292 - private final long term;
293 -
294 - public InnerMutexTask(String mutexPath, MutexTask task, long term) {
295 - this.mutexPath = mutexPath;
296 - this.term = term;
297 - this.task = task;
298 - }
299 -
300 - public long term() {
301 - return term;
302 - }
303 -
304 - @Override
305 - public void start() {
306 - log.debug("Starting execution for mutex task guarded by {}", mutexPath);
307 - task.start();
308 - log.debug("Finished execution for mutex task guarded by {}", mutexPath);
309 - }
310 -
311 - @Override
312 - public void stop() {
313 - log.debug("Stopping execution for mutex task guarded by {}", mutexPath);
314 - task.stop();
315 - }
316 - }
317 -}
...\ No newline at end of file ...\ No newline at end of file
...@@ -94,7 +94,7 @@ public class PartitionManager extends AbstractListenerManager<PartitionEvent, Pa ...@@ -94,7 +94,7 @@ public class PartitionManager extends AbstractListenerManager<PartitionEvent, Pa
94 messagingService, 94 messagingService,
95 clusterService, 95 clusterService,
96 CatalystSerializers.getSerializer(), 96 CatalystSerializers.getSerializer(),
97 - new File(System.getProperty("karaf.data") + "/data/" + partition.getId())))); 97 + new File(System.getProperty("karaf.data") + "/partitions/" + partition.getId()))));
98 98
99 CompletableFuture<Void> openFuture = CompletableFuture.allOf(partitions.values() 99 CompletableFuture<Void> openFuture = CompletableFuture.allOf(partitions.values()
100 .stream() 100 .stream()
......
...@@ -57,7 +57,6 @@ import org.onosproject.store.service.StorageService; ...@@ -57,7 +57,6 @@ import org.onosproject.store.service.StorageService;
57 import org.onosproject.store.service.TransactionContextBuilder; 57 import org.onosproject.store.service.TransactionContextBuilder;
58 import org.slf4j.Logger; 58 import org.slf4j.Logger;
59 59
60 -import com.google.common.collect.ImmutableMap;
61 import com.google.common.collect.Maps; 60 import com.google.common.collect.Maps;
62 import com.google.common.util.concurrent.Futures; 61 import com.google.common.util.concurrent.Futures;
63 62
...@@ -129,7 +128,7 @@ public class StorageManager implements StorageService, StorageAdminService { ...@@ -129,7 +128,7 @@ public class StorageManager implements StorageService, StorageAdminService {
129 @Override 128 @Override
130 public <K, V> ConsistentMapBuilder<K, V> consistentMapBuilder() { 129 public <K, V> ConsistentMapBuilder<K, V> consistentMapBuilder() {
131 checkPermission(STORAGE_WRITE); 130 checkPermission(STORAGE_WRITE);
132 - return new NewDefaultConsistentMapBuilder<>(federatedPrimitiveCreator); 131 + return new DefaultConsistentMapBuilder<>(federatedPrimitiveCreator);
133 } 132 }
134 133
135 @Override 134 @Override
...@@ -141,13 +140,13 @@ public class StorageManager implements StorageService, StorageAdminService { ...@@ -141,13 +140,13 @@ public class StorageManager implements StorageService, StorageAdminService {
141 @Override 140 @Override
142 public <E> DistributedQueueBuilder<E> queueBuilder() { 141 public <E> DistributedQueueBuilder<E> queueBuilder() {
143 checkPermission(STORAGE_WRITE); 142 checkPermission(STORAGE_WRITE);
144 - return new NewDefaultDistributedQueueBuilder<>(federatedPrimitiveCreator); 143 + return new DefaultDistributedQueueBuilder<>(federatedPrimitiveCreator);
145 } 144 }
146 145
147 @Override 146 @Override
148 public AtomicCounterBuilder atomicCounterBuilder() { 147 public AtomicCounterBuilder atomicCounterBuilder() {
149 checkPermission(STORAGE_WRITE); 148 checkPermission(STORAGE_WRITE);
150 - return new NewDefaultAtomicCounterBuilder(federatedPrimitiveCreator); 149 + return new DefaultAtomicCounterBuilder(federatedPrimitiveCreator);
151 } 150 }
152 151
153 @Override 152 @Override
...@@ -163,7 +162,7 @@ public class StorageManager implements StorageService, StorageAdminService { ...@@ -163,7 +162,7 @@ public class StorageManager implements StorageService, StorageAdminService {
163 @Override 162 @Override
164 public TransactionContextBuilder transactionContextBuilder() { 163 public TransactionContextBuilder transactionContextBuilder() {
165 checkPermission(STORAGE_WRITE); 164 checkPermission(STORAGE_WRITE);
166 - return new NewDefaultTransactionContextBuilder(transactionIdGenerator.get(), 165 + return new DefaultTransactionContextBuilder(transactionIdGenerator.get(),
167 federatedPrimitiveCreator, 166 federatedPrimitiveCreator,
168 transactionCoordinator); 167 transactionCoordinator);
169 } 168 }
...@@ -181,26 +180,10 @@ public class StorageManager implements StorageService, StorageAdminService { ...@@ -181,26 +180,10 @@ public class StorageManager implements StorageService, StorageAdminService {
181 180
182 @Override 181 @Override
183 public Map<String, Long> getCounters() { 182 public Map<String, Long> getCounters() {
184 - Map<String, Long> result = Maps.newHashMap();
185 - result.putAll(getInMemoryDatabaseCounters());
186 - result.putAll(getPartitionedDatabaseCounters());
187 - return result;
188 - }
189 -
190 - @Override
191 - public Map<String, Long> getInMemoryDatabaseCounters() {
192 - return ImmutableMap.of();
193 - }
194 -
195 - @Override
196 - public Map<String, Long> getPartitionedDatabaseCounters() {
197 - return getCounters(federatedPrimitiveCreator);
198 - }
199 -
200 - public Map<String, Long> getCounters(DistributedPrimitiveCreator creator) {
201 Map<String, Long> counters = Maps.newConcurrentMap(); 183 Map<String, Long> counters = Maps.newConcurrentMap();
202 - creator.getAsyncAtomicCounterNames() 184 + federatedPrimitiveCreator.getAsyncAtomicCounterNames()
203 - .forEach(name -> counters.put(name, creator.newAsyncCounter(name).asAtomicCounter().get())); 185 + .forEach(name -> counters.put(name,
186 + federatedPrimitiveCreator.newAsyncCounter(name).asAtomicCounter().get()));
204 return counters; 187 return counters;
205 } 188 }
206 189
......
...@@ -35,6 +35,8 @@ import org.onosproject.store.service.Versioned; ...@@ -35,6 +35,8 @@ import org.onosproject.store.service.Versioned;
35 */ 35 */
36 public class UnmodifiableAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMap<K, V> { 36 public class UnmodifiableAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMap<K, V> {
37 37
38 + private static final String ERROR_MSG = "map updates are not allowed";
39 +
38 public UnmodifiableAsyncConsistentMap(AsyncConsistentMap<K, V> backingMap) { 40 public UnmodifiableAsyncConsistentMap(AsyncConsistentMap<K, V> backingMap) {
39 super(backingMap); 41 super(backingMap);
40 } 42 }
...@@ -43,56 +45,56 @@ public class UnmodifiableAsyncConsistentMap<K, V> extends DelegatingAsyncConsist ...@@ -43,56 +45,56 @@ public class UnmodifiableAsyncConsistentMap<K, V> extends DelegatingAsyncConsist
43 public CompletableFuture<Versioned<V>> computeIf(K key, 45 public CompletableFuture<Versioned<V>> computeIf(K key,
44 Predicate<? super V> condition, 46 Predicate<? super V> condition,
45 BiFunction<? super K, ? super V, ? extends V> remappingFunction) { 47 BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
46 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 48 + return Tools.exceptionalFuture(new UnsupportedOperationException(""));
47 } 49 }
48 50
49 @Override 51 @Override
50 public CompletableFuture<Versioned<V>> put(K key, V value) { 52 public CompletableFuture<Versioned<V>> put(K key, V value) {
51 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 53 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
52 } 54 }
53 55
54 @Override 56 @Override
55 public CompletableFuture<Versioned<V>> putAndGet(K key, V value) { 57 public CompletableFuture<Versioned<V>> putAndGet(K key, V value) {
56 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 58 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
57 } 59 }
58 60
59 @Override 61 @Override
60 public CompletableFuture<Versioned<V>> remove(K key) { 62 public CompletableFuture<Versioned<V>> remove(K key) {
61 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 63 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
62 } 64 }
63 65
64 @Override 66 @Override
65 public CompletableFuture<Void> clear() { 67 public CompletableFuture<Void> clear() {
66 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 68 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
67 } 69 }
68 70
69 @Override 71 @Override
70 public CompletableFuture<Versioned<V>> putIfAbsent(K key, V value) { 72 public CompletableFuture<Versioned<V>> putIfAbsent(K key, V value) {
71 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 73 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
72 } 74 }
73 75
74 @Override 76 @Override
75 public CompletableFuture<Boolean> remove(K key, V value) { 77 public CompletableFuture<Boolean> remove(K key, V value) {
76 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 78 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
77 } 79 }
78 80
79 @Override 81 @Override
80 public CompletableFuture<Boolean> remove(K key, long version) { 82 public CompletableFuture<Boolean> remove(K key, long version) {
81 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 83 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
82 } 84 }
83 85
84 @Override 86 @Override
85 public CompletableFuture<Versioned<V>> replace(K key, V value) { 87 public CompletableFuture<Versioned<V>> replace(K key, V value) {
86 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 88 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
87 } 89 }
88 90
89 @Override 91 @Override
90 public CompletableFuture<Boolean> replace(K key, V oldValue, V newValue) { 92 public CompletableFuture<Boolean> replace(K key, V oldValue, V newValue) {
91 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 93 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
92 } 94 }
93 95
94 @Override 96 @Override
95 public CompletableFuture<Boolean> replace(K key, long oldVersion, V newValue) { 97 public CompletableFuture<Boolean> replace(K key, long oldVersion, V newValue) {
96 - return Tools.exceptionalFuture(new UnsupportedOperationException("map updates are not allowed")); 98 + return Tools.exceptionalFuture(new UnsupportedOperationException(ERROR_MSG));
97 } 99 }
98 } 100 }
......
...@@ -15,7 +15,6 @@ ...@@ -15,7 +15,6 @@
15 */ 15 */
16 16
17 /** 17 /**
18 - * Implementation of partitioned and distributed store facility capable of 18 + * Implementation classes for various Distributed primitives.
19 - * providing consistent update semantics.
20 */ 19 */
21 package org.onosproject.store.primitives.impl; 20 package org.onosproject.store.primitives.impl;
...\ No newline at end of file ...\ No newline at end of file
......