Implements accumulation of the fwdobjectives in ofdpa pipelines
Change-Id: I95cbdd9b3fb8d439003a103111a01dc3aee2072b
diff --git a/drivers/default/src/main/java/org/onosproject/driver/pipeline/ofdpa/Ofdpa2Pipeline.java b/drivers/default/src/main/java/org/onosproject/driver/pipeline/ofdpa/Ofdpa2Pipeline.java
index 851b24b..f4ab685 100644
--- a/drivers/default/src/main/java/org/onosproject/driver/pipeline/ofdpa/Ofdpa2Pipeline.java
+++ b/drivers/default/src/main/java/org/onosproject/driver/pipeline/ofdpa/Ofdpa2Pipeline.java
@@ -16,7 +16,9 @@
package org.onosproject.driver.pipeline.ofdpa;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import org.apache.commons.lang3.tuple.Pair;
import org.onlab.osgi.ServiceDirectory;
import org.onlab.packet.EthType;
import org.onlab.packet.Ethernet;
@@ -25,6 +27,8 @@
import org.onlab.packet.IpPrefix;
import org.onlab.packet.MacAddress;
import org.onlab.packet.VlanId;
+import org.onlab.util.AbstractAccumulator;
+import org.onlab.util.Accumulator;
import org.onlab.util.KryoNamespace;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.CoreService;
@@ -43,6 +47,7 @@
import org.onosproject.net.behaviour.PipelinerContext;
import org.onosproject.net.device.DeviceService;
import org.onosproject.net.driver.AbstractHandlerBehaviour;
+import org.onosproject.net.driver.Driver;
import org.onosproject.net.flow.DefaultFlowRule;
import org.onosproject.net.flow.DefaultTrafficSelector;
import org.onosproject.net.flow.DefaultTrafficTreatment;
@@ -98,10 +103,12 @@
import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.Timer;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static java.util.concurrent.Executors.newScheduledThreadPool;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
import static org.onlab.packet.MacAddress.BROADCAST;
import static org.onlab.packet.MacAddress.IPV4_MULTICAST;
import static org.onlab.packet.MacAddress.IPV6_MULTICAST;
@@ -119,6 +126,10 @@
* Driver for Broadcom's OF-DPA v2.0 TTP.
*/
public class Ofdpa2Pipeline extends AbstractHandlerBehaviour implements Pipeliner {
+ // Timer for the accumulator
+ private static final Timer TIMER = new Timer("fwdobj-batching");
+ private Accumulator<Pair<ForwardingObjective, Collection<FlowRule>>> accumulator;
+
protected static final int PORT_TABLE = 0;
protected static final int VLAN_TABLE = 10;
protected static final int VLAN_1_TABLE = 11;
@@ -175,11 +186,15 @@
protected Ofdpa2GroupHandler groupHandler;
// flows installations to be retried
- private ScheduledExecutorService executorService
+ private ScheduledExecutorService retryExecutorService
= newScheduledThreadPool(5, groupedThreads("OfdpaPipeliner", "retry-%d", log));
private static final int MAX_RETRY_ATTEMPTS = 10;
private static final int RETRY_MS = 1000;
+ // accumulator executor service
+ private ScheduledExecutorService accumulatorExecutorService
+ = newSingleThreadScheduledExecutor(groupedThreads("OfdpaPipeliner", "acc-%d", log));
+
@Override
public void init(DeviceId deviceId, PipelinerContext context) {
this.deviceId = deviceId;
@@ -190,6 +205,12 @@
groupService = serviceDirectory.get(GroupService.class);
flowObjectiveStore = context.store();
deviceService = serviceDirectory.get(DeviceService.class);
+ // Init the accumulator, if enabled
+ if (isAccumulatorEnabled()) {
+ accumulator = new ForwardingObjectiveAccumulator(context.accumulatorMaxObjectives(),
+ context.accumulatorMaxBatchMillis(),
+ context.accumulatorMaxIdleMillis());
+ }
initDriverId();
initGroupHander(context);
@@ -213,6 +234,15 @@
// software switches does require table-miss-entries.
}
+ public boolean isAccumulatorEnabled() {
+ Driver driver = super.data().driver();
+ // we cannot determine the property
+ if (driver == null) {
+ return false;
+ }
+ return Boolean.parseBoolean(driver.getProperty(ACCUMULATOR_ENABLED));
+ }
+
/**
* Determines whether this pipeline requires MPLS POP instruction.
*
@@ -329,38 +359,57 @@
// generated by FlowRule service for empty flowOps.
return;
}
- sendForward(fwd, rules);
+ // Let's accumulate flow rules - otherwise send directly
+ if (accumulator != null) {
+ accumulator.add(Pair.of(fwd, rules));
+ } else {
+ sendForwards(Collections.singletonList(Pair.of(fwd, rules)));
+ }
}
- private void sendForward(ForwardingObjective fwd, Collection<FlowRule> rules) {
+ // Builds the batch using the accumulated flow rules
+ private void sendForwards(List<Pair<ForwardingObjective, Collection<FlowRule>>> pairs) {
FlowRuleOperations.Builder flowOpsBuilder = FlowRuleOperations.builder();
- switch (fwd.op()) {
- case ADD:
- rules.stream()
- .filter(Objects::nonNull)
- .forEach(flowOpsBuilder::add);
- log.debug("Applying a add fwd-obj {} to sw:{}", fwd.id(), deviceId);
- break;
- case REMOVE:
- rules.stream()
- .filter(Objects::nonNull)
- .forEach(flowOpsBuilder::remove);
- log.debug("Deleting a flow rule to sw:{}", deviceId);
- break;
- default:
- fail(fwd, ObjectiveError.UNKNOWN);
- log.warn("Unknown forwarding type {}", fwd.op());
- }
-
+ log.debug("Sending {} fwd-objs", pairs.size());
+ List<Objective> fwdObjs = Lists.newArrayList();
+ // Iterates over all accumulated flow rules and then build an unique batch
+ pairs.forEach(pair -> {
+ ForwardingObjective fwd = pair.getLeft();
+ Collection<FlowRule> rules = pair.getRight();
+ switch (fwd.op()) {
+ case ADD:
+ rules.stream()
+ .filter(Objects::nonNull)
+ .forEach(flowOpsBuilder::add);
+ log.debug("Applying a add fwd-obj {} to sw:{}", fwd.id(), deviceId);
+ fwdObjs.add(fwd);
+ break;
+ case REMOVE:
+ rules.stream()
+ .filter(Objects::nonNull)
+ .forEach(flowOpsBuilder::remove);
+ log.debug("Deleting a flow rule to sw:{}", deviceId);
+ fwdObjs.add(fwd);
+ break;
+ default:
+ fail(fwd, ObjectiveError.UNKNOWN);
+ log.warn("Unknown forwarding type {}", fwd.op());
+ }
+ });
+ // Finally applies the operations
flowRuleService.apply(flowOpsBuilder.build(new FlowRuleOperationsContext() {
+
@Override
public void onSuccess(FlowRuleOperations ops) {
- pass(fwd);
+ log.trace("Flow rule operations onSuccess {}", ops);
+ fwdObjs.forEach(Ofdpa2Pipeline::pass);
}
@Override
public void onError(FlowRuleOperations ops) {
- fail(fwd, ObjectiveError.FLOWINSTALLATIONFAILED);
+ ObjectiveError error = ObjectiveError.FLOWINSTALLATIONFAILED;
+ log.warn("Flow rule operations onError {}. Reason = {}", ops, error);
+ fwdObjs.forEach(fwdObj -> fail(fwdObj, error));
}
}));
}
@@ -1528,7 +1577,7 @@
}
if (emptyGroup) {
- executorService.schedule(new RetryFlows(fwd, flowRuleCollection),
+ retryExecutorService.schedule(new RetryFlows(fwd, flowRuleCollection),
RETRY_MS, TimeUnit.MILLISECONDS);
}
return flowRuleCollection;
@@ -1951,9 +2000,42 @@
public void run() {
log.info("RETRY FLOWS ATTEMPT# {} for fwd:{} rules:{}",
MAX_RETRY_ATTEMPTS - attempts, fwd.id(), retryFlows.size());
- sendForward(fwd, retryFlows);
+ sendForwards(Collections.singletonList(Pair.of(fwd, retryFlows)));
if (--attempts > 0) {
- executorService.schedule(this, RETRY_MS, TimeUnit.MILLISECONDS);
+ retryExecutorService.schedule(this, RETRY_MS, TimeUnit.MILLISECONDS);
+ }
+ }
+ }
+
+ // Flow rules accumulator for reducing the number of transactions required to the devices.
+ private final class ForwardingObjectiveAccumulator
+ extends AbstractAccumulator<Pair<ForwardingObjective, Collection<FlowRule>>> {
+
+ ForwardingObjectiveAccumulator(int maxFwd, int maxBatchMS, int maxIdleMS) {
+ super(TIMER, maxFwd, maxBatchMS, maxIdleMS);
+ }
+
+ @Override
+ public void processItems(List<Pair<ForwardingObjective, Collection<FlowRule>>> pairs) {
+ // Triggers creation of a batch using the list of flowrules generated from fwdobjs.
+ accumulatorExecutorService.execute(new FlowRulesBuilderTask(pairs));
+ }
+ }
+
+ // Task for building batch of flow rules in a separate thread.
+ private final class FlowRulesBuilderTask implements Runnable {
+ private final List<Pair<ForwardingObjective, Collection<FlowRule>>> pairs;
+
+ FlowRulesBuilderTask(List<Pair<ForwardingObjective, Collection<FlowRule>>> pairs) {
+ this.pairs = pairs;
+ }
+
+ @Override
+ public void run() {
+ try {
+ sendForwards(pairs);
+ } catch (Exception e) {
+ log.warn("Unable to send forwards", e);
}
}
}