Added a function to the registry to allocate a block of numbers unique within the whole cluster using Curator's DistributedAtomicLong
diff --git a/src/main/java/net/floodlightcontroller/flowcache/FlowManager.java b/src/main/java/net/floodlightcontroller/flowcache/FlowManager.java
index 098e02f..3561e19 100644
--- a/src/main/java/net/floodlightcontroller/flowcache/FlowManager.java
+++ b/src/main/java/net/floodlightcontroller/flowcache/FlowManager.java
@@ -32,13 +32,12 @@
import net.floodlightcontroller.core.module.FloodlightModuleException;
import net.floodlightcontroller.core.module.IFloodlightModule;
import net.floodlightcontroller.core.module.IFloodlightService;
-import net.floodlightcontroller.flowcache.IFlowService;
import net.floodlightcontroller.flowcache.web.FlowWebRoutable;
import net.floodlightcontroller.restserver.IRestApiService;
import net.floodlightcontroller.util.CallerId;
import net.floodlightcontroller.util.DataPath;
-import net.floodlightcontroller.util.Dpid;
import net.floodlightcontroller.util.DataPathEndpoints;
+import net.floodlightcontroller.util.Dpid;
import net.floodlightcontroller.util.FlowEntry;
import net.floodlightcontroller.util.FlowEntryAction;
import net.floodlightcontroller.util.FlowEntryId;
@@ -53,6 +52,8 @@
import net.floodlightcontroller.util.Port;
import net.floodlightcontroller.util.SwitchPort;
import net.onrc.onos.flow.IFlowManager;
+import net.onrc.onos.registry.controller.IControllerRegistryService;
+import net.onrc.onos.registry.controller.IdBlock;
import net.onrc.onos.util.GraphDBConnection;
import net.onrc.onos.util.GraphDBConnection.Transaction;
@@ -63,7 +64,6 @@
import org.openflow.protocol.OFType;
import org.openflow.protocol.action.OFAction;
import org.openflow.protocol.action.OFActionOutput;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -74,6 +74,7 @@
protected IRestApiService restApi;
protected IFloodlightProviderService floodlightProvider;
protected ITopoRouteService topoRouteService;
+ protected IControllerRegistryService registryService;
protected FloodlightModuleContext context;
protected OFMessageDamper messageDamper;
@@ -112,6 +113,9 @@
private BlockingQueue<Runnable> shortestPathQueue = new LinkedBlockingQueue<Runnable>();
private ThreadPoolExecutor shortestPathExecutor =
new ThreadPoolExecutor(10, 10, 5, TimeUnit.SECONDS, shortestPathQueue);
+
+ protected IdBlock idBlock = null;
+ protected long nextId = 0;
class ShortestPathTask implements Runnable {
private int hint;
@@ -656,11 +660,12 @@
@Override
public Collection<Class<? extends IFloodlightService>>
getModuleDependencies() {
- Collection<Class<? extends IFloodlightService>> l =
- new ArrayList<Class<? extends IFloodlightService>>();
- l.add(IFloodlightProviderService.class);
- l.add(ITopoRouteService.class);
- l.add(IRestApiService.class);
+ Collection<Class<? extends IFloodlightService>> l =
+ new ArrayList<Class<? extends IFloodlightService>>();
+ l.add(IFloodlightProviderService.class);
+ l.add(ITopoRouteService.class);
+ l.add(IRestApiService.class);
+ l.add(IControllerRegistryService.class);
return l;
}
@@ -671,6 +676,7 @@
floodlightProvider = context.getServiceImpl(IFloodlightProviderService.class);
topoRouteService = context.getServiceImpl(ITopoRouteService.class);
restApi = context.getServiceImpl(IRestApiService.class);
+ registryService = context.getServiceImpl(IControllerRegistryService.class);
messageDamper = new OFMessageDamper(OFMESSAGE_DAMPER_CAPACITY,
EnumSet.of(OFType.FLOW_MOD),
OFMESSAGE_DAMPER_TIMEOUT);
@@ -699,10 +705,18 @@
@Override
public void startUp(FloodlightModuleContext context) {
- restApi.addRestletRoutable(new FlowWebRoutable());
-
- // Initialize the Flow Entry ID generator
- nextFlowEntryIdPrefix = randomGenerator.nextInt();
+ restApi.addRestletRoutable(new FlowWebRoutable());
+
+ // Initialize the Flow Entry ID generator
+ nextFlowEntryIdPrefix = randomGenerator.nextInt();
+
+ idBlock = registryService.allocateUniqueIdBlock();
+ if (idBlock != null){
+ log.debug("ID block allocated: {}", idBlock);
+ }
+ else{
+ log.error("Null ID block allocated by registry");
+ }
}
/**
diff --git a/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java b/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java
index dadee65..21da47e 100644
--- a/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java
+++ b/src/main/java/net/onrc/onos/registry/controller/IControllerRegistryService.java
@@ -115,4 +115,7 @@
* @return Collection of dpids
*/
public Collection<Long> getSwitchesControlledByController(String controllerId);
+
+ public IdBlock allocateUniqueIdBlock();
+
}
diff --git a/src/main/java/net/onrc/onos/registry/controller/IdBlock.java b/src/main/java/net/onrc/onos/registry/controller/IdBlock.java
new file mode 100644
index 0000000..64b2af8
--- /dev/null
+++ b/src/main/java/net/onrc/onos/registry/controller/IdBlock.java
@@ -0,0 +1,31 @@
+package net.onrc.onos.registry.controller;
+
+public class IdBlock {
+ private long start;
+ private long end;
+ private long size;
+
+ public IdBlock(long start, long end, long size) {
+ this.start = start;
+ this.end = end;
+ this.size = size;
+ }
+
+ public long getStart() {
+ return start;
+ }
+
+ public long getEnd() {
+ return end;
+ }
+
+ public long getSize() {
+ return size;
+ }
+
+ @Override
+ public String toString() {
+ return "IdBlock [start=" + start + ", end=" + end + ", size=" + size
+ + "]";
+ }
+}
diff --git a/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java b/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java
index 2c220fd..e48c519 100644
--- a/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java
+++ b/src/main/java/net/onrc/onos/registry/controller/StandaloneRegistry.java
@@ -115,6 +115,12 @@
String controllerId) {
throw new RuntimeException("Not yet implemented");
}
+
+ @Override
+ public IdBlock allocateUniqueIdBlock(){
+ //XXX Not exactly unique...
+ return new IdBlock(0L, 0x10000000L, 0x10000000L);
+ }
@Override
public Collection<Class<? extends IFloodlightService>> getModuleServices() {
diff --git a/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java b/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
index 76b7ebd..82259a9 100644
--- a/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
+++ b/src/main/java/net/onrc/onos/registry/controller/ZookeeperRegistry.java
@@ -24,6 +24,8 @@
import com.netflix.curator.RetryPolicy;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
+import com.netflix.curator.framework.recipes.atomic.AtomicValue;
+import com.netflix.curator.framework.recipes.atomic.DistributedAtomicLong;
import com.netflix.curator.framework.recipes.cache.ChildData;
import com.netflix.curator.framework.recipes.cache.PathChildrenCache;
import com.netflix.curator.framework.recipes.cache.PathChildrenCache.StartMode;
@@ -33,6 +35,7 @@
import com.netflix.curator.framework.recipes.leader.LeaderLatchEvent;
import com.netflix.curator.framework.recipes.leader.LeaderLatchListener;
import com.netflix.curator.retry.ExponentialBackoffRetry;
+import com.netflix.curator.retry.RetryOneTime;
import com.netflix.curator.x.discovery.ServiceCache;
import com.netflix.curator.x.discovery.ServiceDiscovery;
import com.netflix.curator.x.discovery.ServiceDiscoveryBuilder;
@@ -67,6 +70,10 @@
protected ConcurrentHashMap<String, SwitchLeadershipData> switches;
protected Map<String, PathChildrenCache> switchPathCaches;
+ private final String ID_COUNTER_PATH = "/flowidcounter";
+ private final Long ID_BLOCK_SIZE = 0x100000000L;
+ protected DistributedAtomicLong distributedIdCounter;
+
//Zookeeper performance-related configuration
protected static final int sessionTimeout = 5000;
protected static final int connectionTimeout = 7000;
@@ -372,6 +379,21 @@
return data;
}
+ public IdBlock allocateUniqueIdBlock(){
+ try {
+ AtomicValue<Long> result = null;
+ do {
+ result = distributedIdCounter.add(ID_BLOCK_SIZE);
+ } while (result == null || !result.succeeded());
+
+ return new IdBlock(result.preValue(), result.postValue() - 1, ID_BLOCK_SIZE);
+ } catch (Exception e) {
+ log.error("Error allocating ID block");
+ }
+
+ return null;
+ }
+
/*
* IFloodlightModule
*/
@@ -427,6 +449,10 @@
client.start();
client = client.usingNamespace(namespace);
+ distributedIdCounter = new DistributedAtomicLong(
+ client,
+ ID_COUNTER_PATH,
+ new RetryOneTime(100));
switchCache = new PathChildrenCache(client, switchLatchesPath, true);
switchCache.getListenable().addListener(switchPathCacheListener);
diff --git a/titan/listNotUpdated b/titan/listNotUpdated
new file mode 100644
index 0000000..a44c458
--- /dev/null
+++ b/titan/listNotUpdated
@@ -0,0 +1,4 @@
+g.stopTransaction(SUCCESS)
+g.V('type', 'flow_entry').each{
+if (it.switch_state.equals("FE_SWITCH_NOT_UPDATED")) println it.map.next()
+}
\ No newline at end of file