BUG-3774: 100k flows initial stats fail - fix

 - synchronized all handshake entry points
 - added sweeping barrier after handshake succeeds
 - fixed unit test

Change-Id: I651cd2244c1f58e53503e7af46e19254975d5418
Signed-off-by: Michal Rehak <mirehak@cisco.com>

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/ConnectionContextImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/ConnectionContextImpl.java
index ce6a51e75818c7861c3542d81c02aa3bf4f868e4..3fecb793c5dbb63deb6d838f947b75c12af04e1b 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/ConnectionContextImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/ConnectionContextImpl.java
@@ -29,7 +29,7 @@ import org.slf4j.LoggerFactory;
 public class ConnectionContextImpl implements ConnectionContext {
 
     private final ConnectionAdapter connectionAdapter;
-    private CONNECTION_STATE connectionState;
+    private volatile CONNECTION_STATE connectionState;
     private FeaturesReply featuresReply;
     private NodeId nodeId;
     private DeviceDisconnectedHandler deviceDisconnectedHandler;
@@ -175,17 +175,17 @@ public class ConnectionContextImpl implements ConnectionContext {
     }
 
     @Override
-    public void changeStateToHandshaking() {
+    public synchronized void changeStateToHandshaking() {
         connectionState = CONNECTION_STATE.HANDSHAKING;
     }
 
     @Override
-    public void changeStateToTimeouting() {
+    public synchronized void changeStateToTimeouting() {
         connectionState = CONNECTION_STATE.TIMEOUTING;
     }
 
     @Override
-    public void changeStateToWorking() {
+    public synchronized void changeStateToWorking() {
         connectionState = CONNECTION_STATE.WORKING;
     }
 

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/ConnectionReadyListenerImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/ConnectionReadyListenerImpl.java
index 7c9f0eb6f22225e720e4cd26e375261f71a94bf9..41610e7822b33eb738865f79d1f2c88996a22165 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/ConnectionReadyListenerImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/ConnectionReadyListenerImpl.java
@@ -7,6 +7,7 @@
  */
 package org.opendaylight.openflowplugin.impl.connection.listener;
 
+import java.util.concurrent.Future;
 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
 import org.opendaylight.openflowplugin.api.openflow.connection.HandshakeContext;
@@ -40,12 +41,36 @@ public class ConnectionReadyListenerImpl implements ConnectionReadyListener {
                 connectionContext.getConnectionAdapter().getRemoteAddress());
 
         if (connectionContext.getConnectionState() == null) {
-            HandshakeStepWrapper handshakeStepWrapper = new HandshakeStepWrapper(
-                    null, handshakeContext.getHandshakeManager(), connectionContext.getConnectionAdapter());
-            handshakeContext.getHandshakePool().execute(handshakeStepWrapper);
-            connectionContext.changeStateToHandshaking();
+            synchronized (connectionContext) {
+                if (connectionContext.getConnectionState() == null) {
+                    connectionContext.changeStateToHandshaking();
+                    HandshakeStepWrapper handshakeStepWrapper = new HandshakeStepWrapper(
+                            null, handshakeContext.getHandshakeManager(), connectionContext.getConnectionAdapter());
+                    final Future<?> handshakeResult = handshakeContext.getHandshakePool().submit(handshakeStepWrapper);
+
+                    try {
+                        // as we run not in netty thread, need to remain in sync lock until initial handshake step processed
+                        handshakeResult.get();
+                    } catch (Exception e) {
+                        LOG.warn("failed to process onConnectionReady event on device {}",
+                                connectionContext.getConnectionAdapter().getRemoteAddress(),
+                                e);
+                        connectionContext.closeConnection(false);
+                        try {
+                            handshakeContext.close();
+                        } catch (Exception e1) {
+                            LOG.info("failed to close handshake context for device {}",
+                                    connectionContext.getConnectionAdapter().getRemoteAddress(),
+                                    e1
+                            );
+                        }
+                    }
+                } else {
+                    LOG.debug("already touched by hello message from device {}", connectionContext.getConnectionAdapter().getRemoteAddress());
+                }
+            }
         } else {
-            LOG.debug("already touched by hello message");
+            LOG.debug("already touched by hello message from device {}", connectionContext.getConnectionAdapter().getRemoteAddress());
         }
     }
 

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/HandshakeListenerImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/HandshakeListenerImpl.java
index b17d3675649edacdac47c5f7847b7fa9535381c9..c8f0600edf866903a2f4bfc45545e7c8e304d53e 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/HandshakeListenerImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/HandshakeListenerImpl.java
@@ -8,13 +8,22 @@
 
 package org.opendaylight.openflowplugin.impl.connection.listener;
 
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.JdkFutureAdapters;
+import com.google.common.util.concurrent.ListenableFuture;
+import javax.annotation.Nullable;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
 import org.opendaylight.openflowplugin.api.openflow.connection.HandshakeContext;
 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceConnectedHandler;
 import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeListener;
 import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.SessionStatistics;
 import org.opendaylight.openflowplugin.openflow.md.util.InventoryDataServiceUtil;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierOutput;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
+import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,8 +54,33 @@ public class HandshakeListenerImpl implements HandshakeListener {
         connectionContext.changeStateToWorking();
         connectionContext.setFeatures(featureOutput);
         connectionContext.setNodeId(InventoryDataServiceUtil.nodeIdFromDatapathId(featureOutput.getDatapathId()));
-        deviceConnectedHandler.deviceConnected(connectionContext);
-        SessionStatistics.countEvent(connectionContext.getNodeId().toString(), SessionStatistics.ConnectionStatus.CONNECTION_CREATED);
+
+        // fire barrier in order to sweep all handshake and posthandshake messages before continue
+        final ListenableFuture<RpcResult<BarrierOutput>> barrier = fireBarrier(version, 0L);
+        Futures.addCallback(barrier, new FutureCallback<RpcResult<BarrierOutput>>() {
+            @Override
+            public void onSuccess(@Nullable final RpcResult<BarrierOutput> result) {
+                LOG.debug("succeeded by getting sweep barrier after posthandshake for device {}", connectionContext.getNodeId());
+                deviceConnectedHandler.deviceConnected(connectionContext);
+                SessionStatistics.countEvent(connectionContext.getNodeId().toString(),
+                        SessionStatistics.ConnectionStatus.CONNECTION_CREATED);
+            }
+
+            @Override
+            public void onFailure(final Throwable t) {
+                LOG.info("failed to get sweep barrier after posthandshake for device {}", connectionContext.getNodeId());
+                connectionContext.closeConnection(false);
+            }
+        });
+    }
+
+    protected ListenableFuture<RpcResult<BarrierOutput>> fireBarrier(final Short version, final long xid) {
+        final BarrierInput barrierInput = new BarrierInputBuilder()
+                .setXid(xid)
+                .setVersion(version)
+                .build();
+        return JdkFutureAdapters.listenInPoolThread(
+                connectionContext.getConnectionAdapter().barrier(barrierInput));
     }
 
     @Override

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/OpenflowProtocolListenerInitialImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/OpenflowProtocolListenerInitialImpl.java
index e93cc70ae0088ec862b5e3be0cac7d0a307e92b4..462228863e6d12d322a3eda0aeade8a5999de64a 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/OpenflowProtocolListenerInitialImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/OpenflowProtocolListenerInitialImpl.java
@@ -73,21 +73,28 @@ public class OpenflowProtocolListenerInitialImpl implements OpenflowProtocolList
 
     @Override
     public void onHelloMessage(final HelloMessage hello) {
-        LOG.debug("processing HELLO.xid: {}", hello.getXid());
-        if (connectionContext.getConnectionState() == null) {
-            connectionContext.changeStateToHandshaking();
-        }
-
-        if (checkState(ConnectionContext.CONNECTION_STATE.HANDSHAKING)) {
-            final HandshakeStepWrapper handshakeStepWrapper = new HandshakeStepWrapper(
-                    hello, handshakeContext.getHandshakeManager(), connectionContext.getConnectionAdapter());
-            //handshakeContext.getHandshakePool().submit(handshakeStepWrapper);
-            // use up netty thread
-            handshakeStepWrapper.run();
+        LOG.debug("processing HELLO.xid: {} from device {}", hello.getXid(), connectionContext.getConnectionAdapter().getRemoteAddress());
+        final ConnectionContext.CONNECTION_STATE connectionState = connectionContext.getConnectionState();
+        if (connectionState == null
+                || ConnectionContext.CONNECTION_STATE.HANDSHAKING.equals(connectionState)) {
+            synchronized (connectionContext) {
+                if (connectionContext.getConnectionState() == null) {
+                    // got here before connection ready notification
+                    connectionContext.changeStateToHandshaking();
+                }
+
+                if (checkState(ConnectionContext.CONNECTION_STATE.HANDSHAKING)) {
+                    final HandshakeStepWrapper handshakeStepWrapper = new HandshakeStepWrapper(
+                            hello, handshakeContext.getHandshakeManager(), connectionContext.getConnectionAdapter());
+                    // use up netty thread
+                    handshakeStepWrapper.run();
+                } else {
+                    LOG.debug("already out of handshake phase but still received hello message from device {}", connectionContext.getConnectionAdapter().getRemoteAddress());
+                }
+            }
         } else {
-            //TODO: consider disconnecting of bad behaving device
+            LOG.debug("already touched by onConnectionReady event from device {} (or finished handshake)", connectionContext.getConnectionAdapter().getRemoteAddress());
         }
-
     }
 
     @Override

diff --git a/openflowplugin-impl/src/test/java/org/opendaylight/openflowplugin/impl/connection/ConnectionManagerImplTest.java b/openflowplugin-impl/src/test/java/org/opendaylight/openflowplugin/impl/connection/ConnectionManagerImplTest.java
index e493a0e5f1129ce35f88a40ae76396160ebb3cf6..0f228516affb5d98396a2ce78820b7af0413e737 100644 (file)
--- a/openflowplugin-impl/src/test/java/org/opendaylight/openflowplugin/impl/connection/ConnectionManagerImplTest.java
+++ b/openflowplugin-impl/src/test/java/org/opendaylight/openflowplugin/impl/connection/ConnectionManagerImplTest.java
@@ -7,13 +7,11 @@
  */
 package org.opendaylight.openflowplugin.impl.connection;
 
-import static org.junit.Assert.fail;
 import com.google.common.util.concurrent.SettableFuture;
 import java.math.BigInteger;
 import java.net.InetSocketAddress;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
@@ -27,6 +25,8 @@ import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyList
 import org.opendaylight.openflowplugin.api.OFConstants;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceConnectedHandler;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierOutputBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesInput;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutputBuilder;
@@ -65,6 +65,8 @@ public class ConnectionManagerImplTest {
         final InetSocketAddress deviceAddress = InetSocketAddress.createUnresolved("yahoo", 42);
         Mockito.when(connection.getRemoteAddress()).thenReturn(deviceAddress);
         Mockito.when(connection.isAlive()).thenReturn(true);
+        Mockito.when(connection.barrier(Matchers.<BarrierInput>any()))
+                .thenReturn(RpcResultBuilder.success(new BarrierOutputBuilder().build()).buildFuture());
     }
 
     /**
@@ -177,14 +179,4 @@ public class ConnectionManagerImplTest {
 
         Mockito.verify(deviceConnectedHandler, Mockito.timeout(FINAL_STEP_TIMEOUT)).deviceConnected(Matchers.any(ConnectionContext.class));
     }
-
-    /**
-     * Test method for {@link org.opendaylight.openflowplugin.impl.connection.ConnectionManagerImpl#setOpenflowProtocolListener(org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OpenflowProtocolListener)}.
-     */
-    @Test
-    @Ignore
-    public void testSetOpenflowProtocolListener() {
-        fail("Not yet implemented");
-    }
-
 }