supporting multipart stats replies: tested upto 15K rules

Change-Id: I36fbd99d012a74c1f5240f37b60d3b58be85626c

 package org.onlab.onos.cluster;
 package org.onlab.onos.cluster;
 
 
-import org.onlab.onos.net.MastershipRole;
 import org.onlab.onos.net.provider.Provider;
 import org.onlab.onos.net.provider.Provider;
 
 
 /**
 /**

@@ -150,7 +150,7 @@ public class DefaultFlowRule implements FlowRule {
         if (this == obj) {
         if (this == obj) {
             return true;
             return true;
         }
         }
-        if (obj instanceof FlowRule) {
+        if (obj instanceof DefaultFlowRule) {
             DefaultFlowRule that = (DefaultFlowRule) obj;
             DefaultFlowRule that = (DefaultFlowRule) obj;
             return Objects.equals(deviceId, that.deviceId) &&
             return Objects.equals(deviceId, that.deviceId) &&
                     Objects.equals(id, that.id);
                     Objects.equals(id, that.id);

@@ -138,7 +138,7 @@ implements FlowRuleService, FlowRuleProviderRegistry {
         public void flowMissing(FlowRule flowRule) {
         public void flowMissing(FlowRule flowRule) {
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkValidity();
             checkValidity();
-            log.info("Flow {} has not been installed.", flowRule);
+            log.debug("Flow {} has not been installed.", flowRule);
 
 
         }
         }
 
 
@@ -146,7 +146,7 @@ implements FlowRuleService, FlowRuleProviderRegistry {
         public void extraneousFlow(FlowRule flowRule) {
         public void extraneousFlow(FlowRule flowRule) {
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkValidity();
             checkValidity();
-            log.info("Flow {} is on switch but not in store.", flowRule);
+            log.debug("Flow {} is on switch but not in store.", flowRule);
         }
         }
 
 
         @Override
         @Override
@@ -188,8 +188,10 @@ implements FlowRuleService, FlowRuleProviderRegistry {
                 }
                 }
             }
             }
             for (FlowRule rule : storedRules) {
             for (FlowRule rule : storedRules) {
+
                 // there are rules in the store that aren't on the switch
                 // there are rules in the store that aren't on the switch
                 flowMissing(rule);
                 flowMissing(rule);
+
             }
             }
         }
         }
     }
     }

@@ -79,8 +79,6 @@ public class FlowModBuilder {
                 .setActions(actions)
                 .setActions(actions)
                 .setMatch(match)
                 .setMatch(match)
                 .setFlags(Collections.singleton(OFFlowModFlags.SEND_FLOW_REM))
                 .setFlags(Collections.singleton(OFFlowModFlags.SEND_FLOW_REM))
-                .setIdleTimeout(10)
-                .setHardTimeout(10)
                 .setPriority(priority)
                 .setPriority(priority)
                 .build();
                 .build();
 
 
@@ -98,8 +96,6 @@ public class FlowModBuilder {
                 .setActions(actions)
                 .setActions(actions)
                 .setMatch(match)
                 .setMatch(match)
                 .setFlags(Collections.singleton(OFFlowModFlags.SEND_FLOW_REM))
                 .setFlags(Collections.singleton(OFFlowModFlags.SEND_FLOW_REM))
-                .setIdleTimeout(10)
-                .setHardTimeout(10)
                 .setPriority(priority)
                 .setPriority(priority)
                 .build();
                 .build();
 
 

@@ -2,7 +2,6 @@ package org.onlab.onos.provider.of.flow.impl;
 
 
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.slf4j.LoggerFactory.getLogger;
 
 
-import java.util.List;
 import java.util.Map;
 import java.util.Map;
 
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Activate;
@@ -29,11 +28,13 @@ import org.projectfloodlight.openflow.protocol.OFFlowStatsReply;
 import org.projectfloodlight.openflow.protocol.OFMessage;
 import org.projectfloodlight.openflow.protocol.OFMessage;
 import org.projectfloodlight.openflow.protocol.OFPortStatus;
 import org.projectfloodlight.openflow.protocol.OFPortStatus;
 import org.projectfloodlight.openflow.protocol.OFStatsReply;
 import org.projectfloodlight.openflow.protocol.OFStatsReply;
+import org.projectfloodlight.openflow.protocol.OFStatsReplyFlags;
 import org.projectfloodlight.openflow.protocol.OFStatsType;
 import org.projectfloodlight.openflow.protocol.OFStatsType;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 
 
-import com.google.common.collect.Lists;
+import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Multimap;
 
 
 /**
 /**
  * Provider which uses an OpenFlow controller to detect network
  * Provider which uses an OpenFlow controller to detect network
@@ -114,6 +115,8 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
     implements OpenFlowSwitchListener, OpenFlowEventListener {
     implements OpenFlowSwitchListener, OpenFlowEventListener {
 
 
         private final Map<Dpid, FlowStatsCollector> collectors = Maps.newHashMap();
         private final Map<Dpid, FlowStatsCollector> collectors = Maps.newHashMap();
+        private final Multimap<DeviceId, FlowRule> completeEntries =
+                ArrayListMultimap.create();
 
 
         @Override
         @Override
         public void switchAdded(Dpid dpid) {
         public void switchAdded(Dpid dpid) {
@@ -153,17 +156,24 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
 
 
         }
         }
 
 
-        private void pushFlowMetrics(Dpid dpid, OFStatsReply stats) {
+        private synchronized void pushFlowMetrics(Dpid dpid, OFStatsReply stats) {
             if (stats.getStatsType() != OFStatsType.FLOW) {
             if (stats.getStatsType() != OFStatsType.FLOW) {
                 return;
                 return;
             }
             }
+            DeviceId did = DeviceId.deviceId(Dpid.uri(dpid));
             final OFFlowStatsReply replies = (OFFlowStatsReply) stats;
             final OFFlowStatsReply replies = (OFFlowStatsReply) stats;
-            final List<FlowRule> entries = Lists.newLinkedList();
+            //final List<FlowRule> entries = Lists.newLinkedList();
+
             for (OFFlowStatsEntry reply : replies.getEntries()) {
             for (OFFlowStatsEntry reply : replies.getEntries()) {
-                entries.add(new FlowRuleBuilder(dpid, reply).build());
+                completeEntries.put(did, new FlowRuleBuilder(dpid, reply).build());
+                //entries.add(new FlowRuleBuilder(dpid, reply).build());
+            }
+
+            if (!stats.getFlags().contains(OFStatsReplyFlags.REPLY_MORE)) {
+                log.debug("sending flowstats to core {}", completeEntries.get(did));
+                providerService.pushFlowMetrics(did, completeEntries.get(did));
+                completeEntries.removeAll(did);
             }
             }
-            log.debug("sending flowstats to core {}", entries);
-            providerService.pushFlowMetrics(DeviceId.deviceId(Dpid.uri(dpid)), entries);
         }
         }
 
 
     }
     }