Fixed an issue where consistent map exception in distributed packet store would …

…prevent apps/components from properly starting.

Fixed the fast and net-pingall.xml scenarios.

Change-Id: Ie5712c5c134bb81181dd2bdb307da5ec13851d45

@@ -41,6 +41,7 @@ import org.onosproject.store.cluster.messaging.MessageSubject;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.serializers.KryoSerializer;
 import org.onosproject.store.serializers.KryoSerializer;
 import org.onosproject.store.service.ConsistentMap;
 import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.ConsistentMapException;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.StorageService;
 import org.onosproject.store.service.StorageService;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
@@ -52,6 +53,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 
 import static org.onlab.util.Tools.groupedThreads;
 import static org.onlab.util.Tools.groupedThreads;
+import static org.onlab.util.Tools.retryable;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.slf4j.LoggerFactory.getLogger;
 
 
 /**
 /**
@@ -66,6 +68,8 @@ public class DistributedPacketStore
 
 
     private final Logger log = getLogger(getClass());
     private final Logger log = getLogger(getClass());
 
 
+    private static final int MAX_BACKOFF = 10;
+
     // TODO: make this configurable.
     // TODO: make this configurable.
     private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 4;
     private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 4;
 
 
@@ -159,11 +163,11 @@ public class DistributedPacketStore
         return tracker.requests();
         return tracker.requests();
     }
     }
 
 
-    private class PacketRequestTracker {
+    private final class PacketRequestTracker {
 
 
         private ConsistentMap<TrafficSelector, Set<PacketRequest>> requests;
         private ConsistentMap<TrafficSelector, Set<PacketRequest>> requests;
 
 
-        public PacketRequestTracker() {
+        private PacketRequestTracker() {
             requests = storageService.<TrafficSelector, Set<PacketRequest>>consistentMapBuilder()
             requests = storageService.<TrafficSelector, Set<PacketRequest>>consistentMapBuilder()
                     .withName("onos-packet-requests")
                     .withName("onos-packet-requests")
                     .withPartitionsDisabled()
                     .withPartitionsDisabled()
@@ -171,7 +175,17 @@ public class DistributedPacketStore
                     .build();
                     .build();
         }
         }
 
 
-        public void add(PacketRequest request) {
+        private void add(PacketRequest request) {
+            AtomicBoolean firstRequest =
+                    retryable(this::addInternal, ConsistentMapException.class,
+                              3, MAX_BACKOFF).apply(request);
+            if (firstRequest.get() && delegate != null) {
+                // The instance that makes the first request will push to all devices
+                delegate.requestPackets(request);
+            }
+        }
+
+        private AtomicBoolean addInternal(PacketRequest request) {
             AtomicBoolean firstRequest = new AtomicBoolean(false);
             AtomicBoolean firstRequest = new AtomicBoolean(false);
             requests.compute(request.selector(), (s, existingRequests) -> {
             requests.compute(request.selector(), (s, existingRequests) -> {
                 if (existingRequests == null) {
                 if (existingRequests == null) {
@@ -186,14 +200,20 @@ public class DistributedPacketStore
                     return existingRequests;
                     return existingRequests;
                 }
                 }
             });
             });
+            return firstRequest;
+        }
 
 
-            if (firstRequest.get() && delegate != null) {
-                // The instance that makes the first request will push to all devices
-                delegate.requestPackets(request);
+        private void remove(PacketRequest request) {
+            AtomicBoolean removedLast =
+                    retryable(this::removeInternal, ConsistentMapException.class,
+                              3, MAX_BACKOFF).apply(request);
+            if (removedLast.get() && delegate != null) {
+                // The instance that removes the last request will remove from all devices
+                delegate.cancelPackets(request);
             }
             }
         }
         }
 
 
-        public void remove(PacketRequest request) {
+        private AtomicBoolean removeInternal(PacketRequest request) {
             AtomicBoolean removedLast = new AtomicBoolean(false);
             AtomicBoolean removedLast = new AtomicBoolean(false);
             requests.computeIfPresent(request.selector(), (s, existingRequests) -> {
             requests.computeIfPresent(request.selector(), (s, existingRequests) -> {
                 if (existingRequests.contains(request)) {
                 if (existingRequests.contains(request)) {
@@ -209,15 +229,10 @@ public class DistributedPacketStore
                     return existingRequests;
                     return existingRequests;
                 }
                 }
             });
             });
-
-            if (removedLast.get() && delegate != null) {
-                // The instance that removes the last request will remove from all devices
-                delegate.cancelPackets(request);
-            }
-
+            return removedLast;
         }
         }
 
 
-        public List<PacketRequest> requests() {
+        private List<PacketRequest> requests() {
             List<PacketRequest> list = Lists.newArrayList();
             List<PacketRequest> list = Lists.newArrayList();
             requests.values().forEach(v -> list.addAll(v.value()));
             requests.values().forEach(v -> list.addAll(v.value()));
             list.sort((o1, o2) -> o1.priority().priorityValue() - o2.priority().priorityValue());
             list.sort((o1, o2) -> o1.priority().priorityValue() - o2.priority().priorityValue());

@@ -13,7 +13,7 @@
   ~ See the License for the specific language governing permissions and
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   ~ limitations under the License.
   -->
   -->
-<scenario name="smoke" description="ONOS smoke test">
+<scenario name="fast" description="ONOS fast smoke test">
     <import file="${ONOS_SCENARIOS}/prerequisites.xml"/>
     <import file="${ONOS_SCENARIOS}/prerequisites.xml"/>
 
 
     <import file="${ONOS_SCENARIOS}/setup.xml"/>
     <import file="${ONOS_SCENARIOS}/setup.xml"/>

@@ -22,7 +22,7 @@
               exec="onos-check-apps ${OC1} drivers,openflow,proxyarp,fwd includes"/>
               exec="onos-check-apps ${OC1} drivers,openflow,proxyarp,fwd includes"/>
 
 
         <!-- TODO: take this out when initial pingall sweep is 100% -->
         <!-- TODO: take this out when initial pingall sweep is 100% -->
-        <step name="Initial-Ping-All" requires="Check-Apps"
+        <step name="Initial-Ping-All" requires="Check-Apps" unless="${ONOS_RF_BUG_FIXED}"
               exec="onos-mininet sendAndExpect py net.pingAll(1) --expect received"/>
               exec="onos-mininet sendAndExpect py net.pingAll(1) --expect received"/>
 
 
         <step name="Ping-All-And-Verify" requires="Check-Apps,Initial-Ping-All"
         <step name="Ping-All-And-Verify" requires="Check-Apps,Initial-Ping-All"