commit f1ccee8d599e28e03cbdbaac25173683b8a55bdb
author Yuta HIGUCHI <y-higuchi@onlab.us> Tue Nov 11 20:39:58 2014 -0800
committer Yuta HIGUCHI <y-higuchi@onlab.us> Tue Nov 11 20:40:40 2014 -0800
tree 841f7f656317b57f372cc85b216699328b04df65
parent ecd89261be4ba6f30ff43f7b95aa63c3f32d66fe

FlowRule : handle Future failure and timeouts.

Change-Id: Ie945b7ee936ae48ec3205592c309baebe8538ce0

core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java

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 f65da3f..c94b6c0 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
@@ -85,6 +85,8 @@
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -132,8 +134,7 @@
     private Cache<Integer, SettableFuture<CompletedBatchOperation>> pendingFutures =
             CacheBuilder.newBuilder()
                 .expireAfterWrite(pendingFutureTimeoutMinutes, TimeUnit.MINUTES)
-                // TODO Explicitly fail the future if expired?
-                //.removalListener(listener)
+                .removalListener(new TimeoutFuture())
                 .build();
 
     // Cache of SMaps used for backup data.  each SMap contain device flow table
@@ -541,6 +542,17 @@
         log.debug("removedFromPrimary {}", removed);
     }
 
+    private static final class TimeoutFuture
+        implements RemovalListener<Integer, SettableFuture<CompletedBatchOperation>> {
+        @Override
+        public void onRemoval(RemovalNotification<Integer, SettableFuture<CompletedBatchOperation>> notification) {
+            // wrapping in ExecutionException to support Future.get
+            notification.getValue()
+                .setException(new ExecutionException("Timed out",
+                                                     new TimeoutException()));
+        }
+    }
+
     private final class OnStoreBatch implements ClusterMessageHandler {
         private final NodeId local;
 
@@ -580,7 +592,18 @@
 
                 @Override
                 public void run() {
-                     CompletedBatchOperation result = Futures.getUnchecked(f);
+                    CompletedBatchOperation result;
+                    try {
+                        result = f.get();
+                    } catch (InterruptedException | ExecutionException e) {
+                        log.error("Batch operation failed", e);
+                        // create everything failed response
+                        Set<FlowRule> failures = new HashSet<>(operation.size());
+                        for (FlowRuleBatchEntry op : operation.getOperations()) {
+                            failures.add(op.getTarget());
+                        }
+                        result = new CompletedBatchOperation(false, failures);
+                    }
                     try {
                         message.respond(SERIALIZER.encode(result));
                     } catch (IOException e) {