Cleanup RequestContextStack

[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / rpc / RpcContextImpl.java

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/rpc/RpcContextImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/rpc/RpcContextImpl.java
index d7a5bdcfc8f1b84c90eb94eda6457dba9a20e650..ab39f92f3acd2e1f9e65e2beea69832dd3e86899 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/rpc/RpcContextImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/rpc/RpcContextImpl.java
@@ -7,41 +7,38 @@
  */
 package org.opendaylight.openflowplugin.impl.rpc;
 
-import com.google.common.util.concurrent.SettableFuture;
-import java.util.ArrayList;
 import java.util.Collection;
-import javax.annotation.concurrent.GuardedBy;
+import java.util.HashSet;
+import java.util.concurrent.Semaphore;
 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
+import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeContext;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.opendaylight.yangtools.yang.common.RpcError;
-import org.opendaylight.yangtools.yang.common.RpcResult;
-import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class RpcContextImpl implements RpcContext {
-    private static final Logger LOG = org.slf4j.LoggerFactory.getLogger(RpcContextImpl.class);
-    final RpcProviderRegistry rpcProviderRegistry;
+    private static final Logger LOG = LoggerFactory.getLogger(RpcContextImpl.class);
+    private final RpcProviderRegistry rpcProviderRegistry;
+    private final MessageSpy messageSpy;
+    private final Semaphore tracker;
 
     // TODO: add private Sal salBroker
     private final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier;
-    private final Collection<RoutedRpcRegistration<?>> rpcRegistrations = new ArrayList<>();
+    private final Collection<RoutedRpcRegistration<?>> rpcRegistrations = new HashSet<>();
 
-    @GuardedBy("requestsList")
-    private final Collection<RequestContext<?>> requestsList = new ArrayList<RequestContext<?>>();
-
-    private int maxRequestsPerDevice;
-
-    public RpcContextImpl(final RpcProviderRegistry rpcProviderRegistry, final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier) {
+    public RpcContextImpl(final MessageSpy messageSpy, final RpcProviderRegistry rpcProviderRegistry, final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier, final int maxRequests) {
+        this.messageSpy = messageSpy;
         this.rpcProviderRegistry = rpcProviderRegistry;
         this.nodeInstanceIdentifier = nodeInstanceIdentifier;
+        tracker = new Semaphore(maxRequests, true);
     }
 
     /**
@@ -54,31 +51,7 @@ public class RpcContextImpl implements RpcContext {
         final RoutedRpcRegistration<S> routedRpcReg = rpcProviderRegistry.addRoutedRpcImplementation(serviceClass, serviceInstance);
         routedRpcReg.registerPath(NodeContext.class, nodeInstanceIdentifier);
         rpcRegistrations.add(routedRpcReg);
-        LOG.debug("Registration of service {} for device {}.",serviceClass, nodeInstanceIdentifier);
-    }
-
-    @Override
-    public <T> SettableFuture<RpcResult<T>> storeOrFail(final RequestContext<T> requestContext) {
-        final SettableFuture<RpcResult<T>> rpcResultFuture = requestContext.getFuture();
-
-        final boolean success;
-        // FIXME: use a fixed-size collection, with lockless reserve/set queue
-        synchronized (requestsList) {
-            if (requestsList.size() < maxRequestsPerDevice) {
-                requestsList.add(requestContext);
-                success = true;
-            } else {
-                success = false;
-            }
-        }
-
-        if (!success) {
-            final RpcResult<T> rpcResult = RpcResultBuilder.<T>failed()
-                    .withError(RpcError.ErrorType.APPLICATION, "", "Device's request queue is full.").build();
-            rpcResultFuture.set(rpcResult);
-        }
-
-        return rpcResultFuture;
+        LOG.debug("Registration of service {} for device {}.", serviceClass, nodeInstanceIdentifier);
     }
 
     /**
@@ -87,33 +60,28 @@ public class RpcContextImpl implements RpcContext {
      * @see java.lang.AutoCloseable#close()
      */
     @Override
-    public void close() throws Exception {
+    public void close() {
         for (final RoutedRpcRegistration<?> rpcRegistration : rpcRegistrations) {
             rpcRegistration.unregisterPath(NodeContext.class, nodeInstanceIdentifier);
             rpcRegistration.close();
         }
     }
 
-    /**
-     * @see org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext#setRequestContextQuota(int)
-     */
-    @Override
-    public void setRequestContextQuota(final int maxRequestsPerDevice) {
-        this.maxRequestsPerDevice = maxRequestsPerDevice;
-    }
-
     @Override
-    public <T> void forgetRequestContext(final RequestContext<T> requestContext) {
-        synchronized (requestsList) {
-            requestsList.remove(requestContext);
-            LOG.trace("Removed request context with xid {}. Context request in list {}.",
-                requestContext.getXid().getValue(), requestsList.size());
+    public <T> RequestContext<T> createRequestContext() {
+        if (!tracker.tryAcquire()) {
+            LOG.trace("Device queue {} at capacity", this);
+            return null;
         }
-    }
 
-    @Override
-    public <T> RequestContext<T> createRequestContext() {
-        return new RequestContextImpl<T>(this);
+        return new AbstractRequestContext<T>() {
+            @Override
+            public void close() {
+                tracker.release();
+                LOG.trace("Removed request context with xid {}", getXid().getValue());
+                messageSpy.spyMessage(RpcContextImpl.class, MessageSpy.STATISTIC_GROUP.REQUEST_STACK_FREED);
+            }
+        };
     }
 
     @Override
@@ -121,9 +89,5 @@ public class RpcContextImpl implements RpcContext {
         for (RoutedRpcRegistration<?> registration : rpcRegistrations) {
             registration.close();
         }
-
-        synchronized (requestsList) {
-            requestsList.clear();
-        }
     }
 }