Towards a distributed flow rule store
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java b/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java
index 43fd694..33f1845 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/BatchOperationResult.java
@@ -1,6 +1,6 @@
package org.onlab.onos.net.flow;
-import java.util.List;
+import java.util.Set;
/**
* Interface capturing the result of a batch operation.
@@ -15,9 +15,9 @@
boolean isSuccess();
/**
- * Obtains a list of items which failed.
- * @return a list of failures
+ * Obtains a set of items which failed.
+ * @return a set of failures
*/
- List<T> failedItems();
+ Set<T> failedItems();
}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java b/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java
index e9889cd..4e671e3 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/CompletedBatchOperation.java
@@ -1,18 +1,18 @@
package org.onlab.onos.net.flow;
-import java.util.List;
+import java.util.Set;
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
public class CompletedBatchOperation implements BatchOperationResult<FlowEntry> {
private final boolean success;
- private final List<FlowEntry> failures;
+ private final Set<FlowEntry> failures;
- public CompletedBatchOperation(boolean success, List<FlowEntry> failures) {
+ public CompletedBatchOperation(boolean success, Set<FlowEntry> failures) {
this.success = success;
- this.failures = ImmutableList.copyOf(failures);
+ this.failures = ImmutableSet.copyOf(failures);
}
@Override
@@ -21,7 +21,7 @@
}
@Override
- public List<FlowEntry> failedItems() {
+ public Set<FlowEntry> failedItems() {
return failures;
}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchEvent.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchEvent.java
new file mode 100644
index 0000000..4ba3366
--- /dev/null
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchEvent.java
@@ -0,0 +1,67 @@
+package org.onlab.onos.net.flow;
+
+import org.onlab.onos.event.AbstractEvent;
+
+/**
+ * Describes flow rule batch event.
+ */
+public final class FlowRuleBatchEvent extends AbstractEvent<FlowRuleBatchEvent.Type, FlowRuleBatchRequest> {
+
+ /**
+ * Type of flow rule events.
+ */
+ public enum Type {
+
+ /**
+ * Signifies that a batch operation has been initiated.
+ */
+ BATCH_OPERATION_REQUESTED,
+
+ /**
+ * Signifies that a batch operation has completed.
+ */
+ BATCH_OPERATION_COMPLETED,
+ }
+
+ private final CompletedBatchOperation result;
+
+ /**
+ * Constructs a new FlowRuleBatchEvent.
+ * @param request batch operation request.
+ * @return event.
+ */
+ public static FlowRuleBatchEvent create(FlowRuleBatchRequest request) {
+ FlowRuleBatchEvent event = new FlowRuleBatchEvent(Type.BATCH_OPERATION_REQUESTED, request, null);
+ return event;
+ }
+
+ /**
+ * Constructs a new FlowRuleBatchEvent.
+ * @param request batch operation request.
+ * @param result completed batch operation result.
+ * @return event.
+ */
+ public static FlowRuleBatchEvent create(FlowRuleBatchRequest request, CompletedBatchOperation result) {
+ FlowRuleBatchEvent event = new FlowRuleBatchEvent(Type.BATCH_OPERATION_COMPLETED, request, result);
+ return event;
+ }
+
+ /**
+ * Returns the result of this batch operation.
+ * @return batch operation result.
+ */
+ public CompletedBatchOperation result() {
+ return result;
+ }
+
+ /**
+ * Creates an event of a given type and for the specified flow rule batch.
+ *
+ * @param type flow rule batch event type
+ * @param batch event flow rule batch subject
+ */
+ private FlowRuleBatchEvent(Type type, FlowRuleBatchRequest request, CompletedBatchOperation result) {
+ super(type, request);
+ this.result = result;
+ }
+}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchRequest.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchRequest.java
new file mode 100644
index 0000000..0414fcb
--- /dev/null
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleBatchRequest.java
@@ -0,0 +1,38 @@
+package org.onlab.onos.net.flow;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
+
+import com.google.common.collect.Lists;
+
+public class FlowRuleBatchRequest {
+
+ private final List<FlowEntry> toAdd;
+ private final List<FlowEntry> toRemove;
+
+ public FlowRuleBatchRequest(List<FlowEntry> toAdd, List<FlowEntry> toRemove) {
+ this.toAdd = Collections.unmodifiableList(toAdd);
+ this.toRemove = Collections.unmodifiableList(toRemove);
+ }
+
+ public List<FlowEntry> toAdd() {
+ return toAdd;
+ }
+
+ public List<FlowEntry> toRemove() {
+ return toRemove;
+ }
+
+ public FlowRuleBatchOperation asBatchOperation() {
+ List<FlowRuleBatchEntry> entries = Lists.newArrayList();
+ for (FlowEntry e : toAdd) {
+ entries.add(new FlowRuleBatchEntry(FlowRuleOperation.ADD, e));
+ }
+ for (FlowEntry e : toRemove) {
+ entries.add(new FlowRuleBatchEntry(FlowRuleOperation.REMOVE, e));
+ }
+ return new FlowRuleBatchOperation(entries);
+ }
+}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java
index 3592e39..5a57b88 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java
@@ -1,11 +1,11 @@
package org.onlab.onos.net.flow;
-import java.util.concurrent.Future;
-
import org.onlab.onos.ApplicationId;
import org.onlab.onos.net.intent.BatchOperation;
import org.onlab.onos.net.provider.Provider;
+import com.google.common.util.concurrent.ListenableFuture;
+
/**
* Abstraction of a flow rule provider.
*/
@@ -43,6 +43,6 @@
* @param batch a batch of flow rules
* @return a future indicating the status of this execution
*/
- Future<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch);
+ ListenableFuture<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch);
}
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java
index abb9a10..c53a32d 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java
@@ -1,5 +1,7 @@
package org.onlab.onos.net.flow;
+import java.util.concurrent.Future;
+
import org.onlab.onos.ApplicationId;
import org.onlab.onos.net.DeviceId;
import org.onlab.onos.store.Store;
@@ -7,7 +9,7 @@
/**
* Manages inventory of flow rules; not intended for direct use.
*/
-public interface FlowRuleStore extends Store<FlowRuleEvent, FlowRuleStoreDelegate> {
+public interface FlowRuleStore extends Store<FlowRuleBatchEvent, FlowRuleStoreDelegate> {
/**
* Returns the number of flow rule in the store.
@@ -41,12 +43,26 @@
Iterable<FlowRule> getFlowRulesByAppId(ApplicationId appId);
/**
+ // TODO: Better description of method behavior.
* Stores a new flow rule without generating events.
*
* @param rule the flow rule to add
- * @return true if the rule should be handled locally
*/
- boolean storeFlowRule(FlowRule rule);
+ void storeFlowRule(FlowRule rule);
+
+ /**
+ * Stores a batch of flow rules.
+ * @param batchOperation batch of flow rules.
+ * @return Future response indicating success/failure of the batch operation
+ * all the way down to the device.
+ */
+ Future<CompletedBatchOperation> storeBatch(FlowRuleBatchOperation batchOperation);
+
+ /**
+ * Invoked on the completion of a storeBatch operation.
+ * @param result
+ */
+ void batchOperationComplete(FlowRuleBatchEvent event);
/**
* Marks a flow rule for deletion. Actual deletion will occur
@@ -55,7 +71,7 @@
* @param rule the flow rule to delete
* @return true if the rule should be handled locally
*/
- boolean deleteFlowRule(FlowRule rule);
+ void deleteFlowRule(FlowRule rule);
/**
* Stores a new flow rule, or updates an existing entry.
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java
index 119712b..66973dd 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStoreDelegate.java
@@ -5,5 +5,5 @@
/**
* Flow rule store delegate abstraction.
*/
-public interface FlowRuleStoreDelegate extends StoreDelegate<FlowRuleEvent> {
+public interface FlowRuleStoreDelegate extends StoreDelegate<FlowRuleBatchEvent> {
}
diff --git a/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java b/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
index 67e0867..ba37d22 100644
--- a/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
+++ b/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
@@ -5,8 +5,10 @@
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -30,7 +32,9 @@
import org.onlab.onos.net.flow.FlowRule;
import org.onlab.onos.net.flow.FlowRuleBatchEntry;
import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
+import org.onlab.onos.net.flow.FlowRuleBatchEvent;
import org.onlab.onos.net.flow.FlowRuleBatchOperation;
+import org.onlab.onos.net.flow.FlowRuleBatchRequest;
import org.onlab.onos.net.flow.FlowRuleEvent;
import org.onlab.onos.net.flow.FlowRuleListener;
import org.onlab.onos.net.flow.FlowRuleProvider;
@@ -47,6 +51,9 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
/**
* Provides implementation of the flow NB & SB APIs.
@@ -104,11 +111,7 @@
public void applyFlowRules(FlowRule... flowRules) {
for (int i = 0; i < flowRules.length; i++) {
FlowRule f = flowRules[i];
- boolean local = store.storeFlowRule(f);
- if (local) {
- // TODO: aggregate all local rules and push down once?
- applyFlowRulesToProviders(f);
- }
+ store.storeFlowRule(f);
}
}
@@ -132,11 +135,7 @@
FlowRule f;
for (int i = 0; i < flowRules.length; i++) {
f = flowRules[i];
- boolean local = store.deleteFlowRule(f);
- if (local) {
- // TODO: aggregate all local rules and push down once?
- removeFlowRulesFromProviders(f);
- }
+ store.deleteFlowRule(f);
}
}
@@ -180,33 +179,21 @@
@Override
public Future<CompletedBatchOperation> applyBatch(
FlowRuleBatchOperation batch) {
- Multimap<FlowRuleProvider, FlowRuleBatchEntry> batches =
+ Multimap<DeviceId, FlowRuleBatchEntry> perDeviceBatches =
ArrayListMultimap.create();
List<Future<CompletedBatchOperation>> futures = Lists.newArrayList();
for (FlowRuleBatchEntry fbe : batch.getOperations()) {
final FlowRule f = fbe.getTarget();
- final Device device = deviceService.getDevice(f.deviceId());
- final FlowRuleProvider frp = getProvider(device.providerId());
- batches.put(frp, fbe);
- switch (fbe.getOperator()) {
- case ADD:
- store.storeFlowRule(f);
- break;
- case REMOVE:
- store.deleteFlowRule(f);
- break;
- case MODIFY:
- default:
- log.error("Batch operation type {} unsupported.", fbe.getOperator());
- }
+ perDeviceBatches.put(f.deviceId(), fbe);
}
- for (FlowRuleProvider provider : batches.keySet()) {
+
+ for (DeviceId deviceId : perDeviceBatches.keySet()) {
FlowRuleBatchOperation b =
- new FlowRuleBatchOperation(batches.get(provider));
- Future<CompletedBatchOperation> future = provider.executeBatch(b);
+ new FlowRuleBatchOperation(perDeviceBatches.get(deviceId));
+ Future<CompletedBatchOperation> future = store.storeBatch(b);
futures.add(future);
}
- return new FlowRuleBatchFuture(futures, batches);
+ return new FlowRuleBatchFuture(futures, perDeviceBatches);
}
@Override
@@ -318,6 +305,7 @@
post(event);
}
} else {
+ log.info("Removing flow rules....");
removeFlowRules(flowEntry);
}
@@ -385,21 +373,47 @@
// Store delegate to re-post events emitted from the store.
private class InternalStoreDelegate implements FlowRuleStoreDelegate {
+ // TODO: Right now we only dispatch events at individual flowEntry level.
+ // It may be more efficient for also dispatch events as a batch.
@Override
- public void notify(FlowRuleEvent event) {
+ public void notify(FlowRuleBatchEvent event) {
+ final FlowRuleBatchRequest request = event.subject();
switch (event.type()) {
- case RULE_ADD_REQUESTED:
- applyFlowRulesToProviders(event.subject());
- break;
- case RULE_REMOVE_REQUESTED:
- removeFlowRulesFromProviders(event.subject());
- break;
+ case BATCH_OPERATION_REQUESTED:
+// for (FlowEntry entry : request.toAdd()) {
+// //eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADD_REQUESTED, entry));
+// }
+// for (FlowEntry entry : request.toRemove()) {
+// //eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVE_REQUESTED, entry));
+// }
+// // FIXME: what about op.equals(FlowRuleOperation.MODIFY) ?
+//
+ FlowRuleBatchOperation batchOperation = request.asBatchOperation();
- case RULE_ADDED:
- case RULE_REMOVED:
- case RULE_UPDATED:
- // only dispatch events related to switch
- eventDispatcher.post(event);
+ FlowRuleProvider flowRuleProvider =
+ getProvider(batchOperation.getOperations().get(0).getTarget().deviceId());
+ final ListenableFuture<CompletedBatchOperation> result =
+ flowRuleProvider.executeBatch(batchOperation);
+ result.addListener(new Runnable() {
+ @Override
+ public void run() {
+ store.batchOperationComplete(FlowRuleBatchEvent.create(request, Futures.getUnchecked(result)));
+ }
+ }, Executors.newCachedThreadPool());
+
+ break;
+ case BATCH_OPERATION_COMPLETED:
+ Set<FlowEntry> failedItems = event.result().failedItems();
+ for (FlowEntry entry : request.toAdd()) {
+ if (!failedItems.contains(entry)) {
+ eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADDED, entry));
+ }
+ }
+ for (FlowEntry entry : request.toRemove()) {
+ if (!failedItems.contains(entry)) {
+ eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVED, entry));
+ }
+ }
break;
default:
break;
@@ -407,18 +421,15 @@
}
}
- private class FlowRuleBatchFuture
- implements Future<CompletedBatchOperation> {
+ private class FlowRuleBatchFuture implements Future<CompletedBatchOperation> {
private final List<Future<CompletedBatchOperation>> futures;
- private final Multimap<FlowRuleProvider, FlowRuleBatchEntry> batches;
+ private final Multimap<DeviceId, FlowRuleBatchEntry> batches;
private final AtomicReference<BatchState> state;
private CompletedBatchOperation overall;
-
-
public FlowRuleBatchFuture(List<Future<CompletedBatchOperation>> futures,
- Multimap<FlowRuleProvider, FlowRuleBatchEntry> batches) {
+ Multimap<DeviceId, FlowRuleBatchEntry> batches) {
this.futures = futures;
this.batches = batches;
state = new AtomicReference<FlowRuleManager.BatchState>();
@@ -460,7 +471,7 @@
}
boolean success = true;
- List<FlowEntry> failed = Lists.newLinkedList();
+ Set<FlowEntry> failed = Sets.newHashSet();
CompletedBatchOperation completed;
for (Future<CompletedBatchOperation> future : futures) {
completed = future.get();
@@ -480,7 +491,7 @@
return overall;
}
boolean success = true;
- List<FlowEntry> failed = Lists.newLinkedList();
+ Set<FlowEntry> failed = Sets.newHashSet();
CompletedBatchOperation completed;
long start = System.nanoTime();
long end = start + unit.toNanos(timeout);
@@ -494,7 +505,7 @@
return finalizeBatchOperation(success, failed);
}
- private boolean validateBatchOperation(List<FlowEntry> failed,
+ private boolean validateBatchOperation(Set<FlowEntry> failed,
CompletedBatchOperation completed) {
if (isCancelled()) {
@@ -516,7 +527,7 @@
}
private CompletedBatchOperation finalizeBatchOperation(boolean success,
- List<FlowEntry> failed) {
+ Set<FlowEntry> failed) {
synchronized (this) {
if (!state.compareAndSet(BatchState.STARTED, BatchState.FINISHED)) {
if (state.get() == BatchState.FINISHED) {
@@ -539,11 +550,6 @@
store.storeFlowRule(fbe.getTarget());
}
}
-
}
}
-
-
-
-
}
diff --git a/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java b/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
index ca7cc07..59b2963 100644
--- a/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
+++ b/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
@@ -1,6 +1,5 @@
package org.onlab.onos.net.flow.impl;
-import static java.util.Collections.EMPTY_LIST;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -17,6 +16,7 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -64,6 +64,7 @@
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ListenableFuture;
/**
* Test codifying the flow rule service & flow rule provider service contracts.
@@ -515,13 +516,13 @@
}
@Override
- public Future<CompletedBatchOperation> executeBatch(
+ public ListenableFuture<CompletedBatchOperation> executeBatch(
BatchOperation<FlowRuleBatchEntry> batch) {
return new TestInstallationFuture();
}
private class TestInstallationFuture
- implements Future<CompletedBatchOperation> {
+ implements ListenableFuture<CompletedBatchOperation> {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
@@ -541,7 +542,7 @@
@Override
public CompletedBatchOperation get()
throws InterruptedException, ExecutionException {
- return new CompletedBatchOperation(true, EMPTY_LIST);
+ return new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet());
}
@Override
@@ -550,6 +551,11 @@
ExecutionException, TimeoutException {
return null;
}
+
+ @Override
+ public void addListener(Runnable task, Executor executor) {
+ // TODO: add stuff.
+ }
}
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
index bde57c6..85f928a 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
@@ -5,10 +5,14 @@
import static org.onlab.onos.store.flow.impl.FlowStoreMessageSubjects.*;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.List;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
@@ -19,11 +23,17 @@
import org.onlab.onos.ApplicationId;
import org.onlab.onos.cluster.ClusterService;
import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.flow.CompletedBatchOperation;
import org.onlab.onos.net.flow.DefaultFlowEntry;
import org.onlab.onos.net.flow.FlowEntry;
import org.onlab.onos.net.flow.FlowEntry.FlowEntryState;
import org.onlab.onos.net.flow.FlowRule;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry;
+import org.onlab.onos.net.flow.FlowRuleBatchEvent;
+import org.onlab.onos.net.flow.FlowRuleBatchOperation;
+import org.onlab.onos.net.flow.FlowRuleBatchRequest;
import org.onlab.onos.net.flow.FlowRuleEvent;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
import org.onlab.onos.net.flow.FlowRuleEvent.Type;
import org.onlab.onos.net.flow.FlowRuleStore;
import org.onlab.onos.net.flow.FlowRuleStoreDelegate;
@@ -43,6 +53,7 @@
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Multimap;
+import com.google.common.util.concurrent.Futures;
/**
* Manages inventory of flow rules using a distributed state management protocol.
@@ -50,7 +61,7 @@
@Component(immediate = true)
@Service
public class DistributedFlowRuleStore
- extends AbstractStore<FlowRuleEvent, FlowRuleStoreDelegate>
+ extends AbstractStore<FlowRuleBatchEvent, FlowRuleStoreDelegate>
implements FlowRuleStore {
private final Logger log = getLogger(getClass());
@@ -92,7 +103,7 @@
public void handle(ClusterMessage message) {
FlowRule rule = SERIALIZER.decode(message.payload());
log.info("received add request for {}", rule);
- storeFlowEntryInternal(rule);
+ storeFlowRule(rule);
// FIXME what to respond.
try {
message.respond(SERIALIZER.encode("ACK"));
@@ -108,7 +119,7 @@
public void handle(ClusterMessage message) {
FlowRule rule = SERIALIZER.decode(message.payload());
log.info("received delete request for {}", rule);
- deleteFlowRuleInternal(rule);
+ deleteFlowRule(rule);
// FIXME what to respond.
try {
message.respond(SERIALIZER.encode("ACK"));
@@ -118,6 +129,22 @@
}
});
+
+ clusterCommunicator.addSubscriber(GET_FLOW_ENTRY, new ClusterMessageHandler() {
+
+ @Override
+ public void handle(ClusterMessage message) {
+ FlowRule rule = SERIALIZER.decode(message.payload());
+ log.info("received get flow entry request for {}", rule);
+ FlowEntry flowEntry = getFlowEntryInternal(rule);
+ try {
+ message.respond(SERIALIZER.encode(flowEntry));
+ } catch (IOException e) {
+ log.error("Failed to respond back", e);
+ }
+ }
+ });
+
log.info("Started");
}
@@ -127,6 +154,9 @@
}
+ // TODO: This is not a efficient operation on a distributed sharded
+ // flow store. We need to revisit the need for this operation or at least
+ // make it device specific.
@Override
public int getFlowRuleCount() {
return flowEntries.size();
@@ -134,7 +164,26 @@
@Override
public synchronized FlowEntry getFlowEntry(FlowRule rule) {
- return getFlowEntryInternal(rule);
+ ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(rule.deviceId());
+ if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
+ return getFlowEntryInternal(rule);
+ }
+
+ log.info("Forwarding getFlowEntry to {}, which is the primary (master) for device {}",
+ replicaInfo.master().orNull(), rule.deviceId());
+
+ ClusterMessage message = new ClusterMessage(
+ clusterService.getLocalNode().id(),
+ FlowStoreMessageSubjects.GET_FLOW_ENTRY,
+ SERIALIZER.encode(rule));
+
+ try {
+ ClusterMessageResponse response = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
+ return SERIALIZER.decode(response.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
+ } catch (IOException | TimeoutException e) {
+ // FIXME: throw a FlowStoreException
+ throw new RuntimeException(e);
+ }
}
private synchronized StoredFlowEntry getFlowEntryInternal(FlowRule rule) {
@@ -165,19 +214,30 @@
}
@Override
- public boolean storeFlowRule(FlowRule rule) {
- ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(rule.deviceId());
- if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
- return storeFlowEntryInternal(rule);
+ public void storeFlowRule(FlowRule rule) {
+ storeBatch(new FlowRuleBatchOperation(Arrays.asList(new FlowRuleBatchEntry(FlowRuleOperation.ADD, rule))));
+ }
+
+ public Future<CompletedBatchOperation> storeBatch(FlowRuleBatchOperation operation) {
+ if (operation.getOperations().isEmpty()) {
+ return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
}
- log.info("Forwarding storeFlowRule to {}, which is the primary (master) for device {}",
- replicaInfo.master().orNull(), rule.deviceId());
+ DeviceId deviceId = operation.getOperations().get(0).getTarget().deviceId();
+
+ ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(deviceId);
+
+ if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
+ return storeBatchInternal(operation);
+ }
+
+ log.info("Forwarding storeBatch to {}, which is the primary (master) for device {}",
+ replicaInfo.master().orNull(), deviceId);
ClusterMessage message = new ClusterMessage(
clusterService.getLocalNode().id(),
FlowStoreMessageSubjects.STORE_FLOW_RULE,
- SERIALIZER.encode(rule));
+ SERIALIZER.encode(operation));
try {
ClusterMessageResponse response = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
@@ -186,58 +246,44 @@
// FIXME: throw a FlowStoreException
throw new RuntimeException(e);
}
- return false;
+
+ return null;
}
- private synchronized boolean storeFlowEntryInternal(FlowRule flowRule) {
- StoredFlowEntry flowEntry = new DefaultFlowEntry(flowRule);
- DeviceId deviceId = flowRule.deviceId();
- // write to local copy.
- if (!flowEntries.containsEntry(deviceId, flowEntry)) {
- flowEntries.put(deviceId, flowEntry);
- flowEntriesById.put(flowRule.appId(), flowEntry);
- notifyDelegate(new FlowRuleEvent(Type.RULE_ADD_REQUESTED, flowRule));
- return true;
+ private Future<CompletedBatchOperation> storeBatchInternal(FlowRuleBatchOperation operation) {
+ List<FlowEntry> toRemove = new ArrayList<>();
+ List<FlowEntry> toAdd = new ArrayList<>();
+ // TODO: backup changes to hazelcast map
+ for (FlowRuleBatchEntry batchEntry : operation.getOperations()) {
+ FlowRule flowRule = batchEntry.getTarget();
+ FlowRuleOperation op = batchEntry.getOperator();
+ if (op.equals(FlowRuleOperation.REMOVE)) {
+ StoredFlowEntry entry = getFlowEntryInternal(flowRule);
+ if (entry != null) {
+ entry.setState(FlowEntryState.PENDING_REMOVE);
+ }
+ toRemove.add(entry);
+ } else if (op.equals(FlowRuleOperation.ADD)) {
+ StoredFlowEntry flowEntry = new DefaultFlowEntry(flowRule);
+ DeviceId deviceId = flowRule.deviceId();
+ if (!flowEntries.containsEntry(deviceId, flowEntry)) {
+ flowEntries.put(deviceId, flowEntry);
+ flowEntriesById.put(flowRule.appId(), flowEntry);
+ toAdd.add(flowEntry);
+ }
+ }
}
- // write to backup.
- // TODO: write to a hazelcast map.
- return false;
+ if (toAdd.isEmpty() && toRemove.isEmpty()) {
+ return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
+ }
+ notifyDelegate(FlowRuleBatchEvent.create(new FlowRuleBatchRequest(toAdd, toRemove)));
+ // TODO: imlpement this.
+ return Futures.immediateFailedFuture(new RuntimeException("Implement this."));
}
@Override
- public synchronized boolean deleteFlowRule(FlowRule rule) {
- ReplicaInfo replicaInfo = replicaInfoManager.getReplicaInfoFor(rule.deviceId());
- if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
- return deleteFlowRuleInternal(rule);
- }
-
- ClusterMessage message = new ClusterMessage(
- clusterService.getLocalNode().id(),
- FlowStoreMessageSubjects.DELETE_FLOW_RULE,
- SERIALIZER.encode(rule));
-
- try {
- ClusterMessageResponse response = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
- response.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
- } catch (IOException | TimeoutException e) {
- // FIXME: throw a FlowStoreException
- throw new RuntimeException(e);
- }
- return false;
- }
-
- private synchronized boolean deleteFlowRuleInternal(FlowRule flowRule) {
- StoredFlowEntry entry = getFlowEntryInternal(flowRule);
- if (entry == null) {
- return false;
- }
- entry.setState(FlowEntryState.PENDING_REMOVE);
-
- // TODO: also update backup.
-
- notifyDelegate(new FlowRuleEvent(Type.RULE_REMOVE_REQUESTED, flowRule));
-
- return true;
+ public void deleteFlowRule(FlowRule rule) {
+ storeBatch(new FlowRuleBatchOperation(Arrays.asList(new FlowRuleBatchEntry(FlowRuleOperation.REMOVE, rule))));
}
@Override
@@ -315,4 +361,9 @@
}
// TODO: also update backup.
}
+
+ @Override
+ public void batchOperationComplete(FlowRuleBatchEvent event) {
+ notifyDelegate(event);
+ }
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java
index a43dad6..ca833b8 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/FlowStoreMessageSubjects.java
@@ -12,4 +12,5 @@
public static final MessageSubject ADD_OR_UPDATE_FLOW_RULE =
new MessageSubject("peer-forward-add-or-update-flow-rule");
public static final MessageSubject REMOVE_FLOW_RULE = new MessageSubject("peer-forward-remove-flow-rule");
+ public static final MessageSubject GET_FLOW_ENTRY = new MessageSubject("peer-forward-get-flow-entry");
}
diff --git a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java
index 9de6d8c..e6259d8 100644
--- a/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java
+++ b/core/store/serializers/src/main/java/org/onlab/onos/store/serializers/KryoNamespaces.java
@@ -26,10 +26,12 @@
import org.onlab.onos.net.PortNumber;
import org.onlab.onos.net.device.DefaultDeviceDescription;
import org.onlab.onos.net.device.DefaultPortDescription;
+import org.onlab.onos.net.flow.DefaultFlowEntry;
import org.onlab.onos.net.flow.DefaultFlowRule;
import org.onlab.onos.net.flow.DefaultTrafficSelector;
import org.onlab.onos.net.flow.DefaultTrafficTreatment;
import org.onlab.onos.net.flow.FlowId;
+import org.onlab.onos.net.flow.StoredFlowEntry;
import org.onlab.onos.net.flow.criteria.Criteria;
import org.onlab.onos.net.flow.criteria.Criterion;
import org.onlab.onos.net.flow.instructions.Instructions;
@@ -93,6 +95,8 @@
HostId.class,
HostDescription.class,
DefaultHostDescription.class,
+ DefaultFlowEntry.class,
+ StoredFlowEntry.class,
DefaultFlowRule.class,
FlowId.class,
DefaultTrafficSelector.class,
diff --git a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
index a96cacb..3d10d3d 100644
--- a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
+++ b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
@@ -3,6 +3,8 @@
import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
import static org.slf4j.LoggerFactory.getLogger;
import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
+
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -10,6 +12,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
@@ -17,11 +20,17 @@
import org.apache.felix.scr.annotations.Service;
import org.onlab.onos.ApplicationId;
import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.flow.CompletedBatchOperation;
import org.onlab.onos.net.flow.DefaultFlowEntry;
import org.onlab.onos.net.flow.FlowEntry;
import org.onlab.onos.net.flow.FlowEntry.FlowEntryState;
import org.onlab.onos.net.flow.FlowId;
import org.onlab.onos.net.flow.FlowRule;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry;
+import org.onlab.onos.net.flow.FlowRuleBatchEntry.FlowRuleOperation;
+import org.onlab.onos.net.flow.FlowRuleBatchEvent;
+import org.onlab.onos.net.flow.FlowRuleBatchOperation;
+import org.onlab.onos.net.flow.FlowRuleBatchRequest;
import org.onlab.onos.net.flow.FlowRuleEvent;
import org.onlab.onos.net.flow.FlowRuleEvent.Type;
import org.onlab.onos.net.flow.FlowRuleStore;
@@ -33,6 +42,7 @@
import com.google.common.base.Function;
import com.google.common.collect.FluentIterable;
+import com.google.common.util.concurrent.Futures;
/**
* Manages inventory of flow rules using trivial in-memory implementation.
@@ -40,7 +50,7 @@
@Component(immediate = true)
@Service
public class SimpleFlowRuleStore
- extends AbstractStore<FlowRuleEvent, FlowRuleStoreDelegate>
+ extends AbstractStore<FlowRuleBatchEvent, FlowRuleStoreDelegate>
implements FlowRuleStore {
private final Logger log = getLogger(getClass());
@@ -148,12 +158,11 @@
}
@Override
- public boolean storeFlowRule(FlowRule rule) {
- final boolean added = storeFlowRuleInternal(rule);
- return added;
+ public void storeFlowRule(FlowRule rule) {
+ storeFlowRuleInternal(rule);
}
- private boolean storeFlowRuleInternal(FlowRule rule) {
+ private void storeFlowRuleInternal(FlowRule rule) {
StoredFlowEntry f = new DefaultFlowEntry(rule);
final DeviceId did = f.deviceId();
final FlowId fid = f.id();
@@ -162,19 +171,20 @@
for (StoredFlowEntry fe : existing) {
if (fe.equals(rule)) {
// was already there? ignore
- return false;
+ return;
}
}
// new flow rule added
existing.add(f);
- // TODO: Should we notify only if it's "remote" event?
- //notifyDelegate(new FlowRuleEvent(Type.RULE_ADD_REQUESTED, rule));
- return true;
+ notifyDelegate(FlowRuleBatchEvent.create(
+ new FlowRuleBatchRequest(
+ Arrays.<FlowEntry>asList(f),
+ Collections.<FlowEntry>emptyList())));
}
}
@Override
- public boolean deleteFlowRule(FlowRule rule) {
+ public void deleteFlowRule(FlowRule rule) {
List<StoredFlowEntry> entries = getFlowEntries(rule.deviceId(), rule.id());
synchronized (entries) {
@@ -184,13 +194,11 @@
entry.setState(FlowEntryState.PENDING_REMOVE);
// TODO: Should we notify only if it's "remote" event?
//notifyDelegate(new FlowRuleEvent(Type.RULE_REMOVE_REQUESTED, rule));
- return true;
}
}
}
}
//log.warn("Cannot find rule {}", rule);
- return false;
}
@Override
@@ -236,4 +244,24 @@
}
return null;
}
+
+ @Override
+ public Future<CompletedBatchOperation> storeBatch(
+ FlowRuleBatchOperation batchOperation) {
+ for (FlowRuleBatchEntry entry : batchOperation.getOperations()) {
+ if (entry.getOperator().equals(FlowRuleOperation.ADD)) {
+ storeFlowRule(entry.getTarget());
+ } else if (entry.getOperator().equals(FlowRuleOperation.REMOVE)) {
+ deleteFlowRule(entry.getTarget());
+ } else {
+ throw new UnsupportedOperationException("Unsupported operation type");
+ }
+ }
+ return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
+ }
+
+ @Override
+ public void batchOperationComplete(FlowRuleBatchEvent event) {
+ notifyDelegate(event);
+ }
}
diff --git a/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java b/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java
index 54265ba..b7d84f0 100644
--- a/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java
+++ b/providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java
@@ -10,7 +10,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -69,9 +69,11 @@
import org.slf4j.Logger;
import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ExecutionList;
+import com.google.common.util.concurrent.ListenableFuture;
/**
* Provider which uses an OpenFlow controller to detect network
@@ -97,6 +99,8 @@
private final InternalFlowProvider listener = new InternalFlowProvider();
+ // FIXME: This should be an expiring map to ensure futures that don't have
+ // a future eventually get garbage collected.
private final Map<Long, InstallationFuture> pendingFutures =
new ConcurrentHashMap<Long, InstallationFuture>();
@@ -159,7 +163,7 @@
}
@Override
- public Future<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch) {
+ public ListenableFuture<CompletedBatchOperation> executeBatch(BatchOperation<FlowRuleBatchEntry> batch) {
final Set<Dpid> sws =
Collections.newSetFromMap(new ConcurrentHashMap<Dpid, Boolean>());
final Map<Long, FlowRuleBatchEntry> fmXids = new HashMap<Long, FlowRuleBatchEntry>();
@@ -315,18 +319,20 @@
}
}
- private class InstallationFuture implements Future<CompletedBatchOperation> {
+ private class InstallationFuture implements ListenableFuture<CompletedBatchOperation> {
private final Set<Dpid> sws;
private final AtomicBoolean ok = new AtomicBoolean(true);
private final Map<Long, FlowRuleBatchEntry> fms;
- private final List<FlowEntry> offendingFlowMods = Lists.newLinkedList();
+ private final Set<FlowEntry> offendingFlowMods = Sets.newHashSet();
private final CountDownLatch countDownLatch;
private Long pendingXid;
private BatchState state;
+ private final ExecutionList executionList = new ExecutionList();
+
public InstallationFuture(Set<Dpid> sws, Map<Long, FlowRuleBatchEntry> fmXids) {
this.state = BatchState.STARTED;
this.sws = sws;
@@ -335,6 +341,7 @@
}
public void fail(OFErrorMsg msg, Dpid dpid) {
+
ok.set(false);
removeRequirement(dpid);
FlowEntry fe = null;
@@ -407,6 +414,9 @@
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
+ if (isDone()) {
+ return false;
+ }
ok.set(false);
this.state = BatchState.CANCELLED;
cleanUp();
@@ -419,7 +429,8 @@
}
}
- return isCancelled();
+ invokeCallbacks();
+ return true;
}
@Override
@@ -429,14 +440,15 @@
@Override
public boolean isDone() {
- return this.state == BatchState.FINISHED;
+ return this.state == BatchState.FINISHED || isCancelled();
}
@Override
public CompletedBatchOperation get() throws InterruptedException, ExecutionException {
countDownLatch.await();
this.state = BatchState.FINISHED;
- return new CompletedBatchOperation(ok.get(), offendingFlowMods);
+ CompletedBatchOperation result = new CompletedBatchOperation(ok.get(), offendingFlowMods);
+ return result;
}
@Override
@@ -445,7 +457,8 @@
TimeoutException {
if (countDownLatch.await(timeout, unit)) {
this.state = BatchState.FINISHED;
- return new CompletedBatchOperation(ok.get(), offendingFlowMods);
+ CompletedBatchOperation result = new CompletedBatchOperation(ok.get(), offendingFlowMods);
+ return result;
}
throw new TimeoutException();
}
@@ -463,10 +476,21 @@
private void removeRequirement(Dpid dpid) {
countDownLatch.countDown();
+ if (countDownLatch.getCount() == 0) {
+ invokeCallbacks();
+ }
sws.remove(dpid);
cleanUp();
}
+ @Override
+ public void addListener(Runnable runnable, Executor executor) {
+ executionList.add(runnable, executor);
+ }
+
+ private void invokeCallbacks() {
+ executionList.execute();
+ }
}
}