X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fopenflow%2Fmd%2Fcore%2FConnectionConductorImpl.java;h=210c88adb768683df90039b55bbeb7476c2f6a4f;hb=da38c1d9f1fc59c08fb80fa4389a900e08c69982;hp=f67fe16d512f2d707416cb80ae06c4dbc903d5d1;hpb=b15e1cd85ee17a2209bbe85da53ed3dcad03bd4b;p=openflowplugin.git diff --git a/openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/core/ConnectionConductorImpl.java b/openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/core/ConnectionConductorImpl.java index f67fe16d51..210c88adb7 100644 --- a/openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/core/ConnectionConductorImpl.java +++ b/openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/core/ConnectionConductorImpl.java @@ -1,5 +1,5 @@ /** - * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved. + * Copyright (c) 2013-2014 Cisco Systems, Inc. and others. All rights reserved. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, @@ -8,19 +8,30 @@ package org.opendaylight.openflowplugin.openflow.md.core; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter; import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener; -import org.opendaylight.openflowplugin.openflow.md.OFConstants; +import org.opendaylight.openflowplugin.api.OFConstants; +import org.opendaylight.openflowplugin.api.openflow.md.core.ConnectionConductor; +import org.opendaylight.openflowplugin.api.openflow.md.core.ErrorHandler; +import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeListener; +import org.opendaylight.openflowplugin.api.openflow.md.core.NotificationEnqueuer; +import org.opendaylight.openflowplugin.api.openflow.md.core.NotificationQueueWrapper; +import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher; +import org.opendaylight.openflowplugin.api.openflow.md.core.session.SessionContext; +import org.opendaylight.openflowplugin.api.openflow.md.core.session.SessionManager; +import org.opendaylight.openflowplugin.api.openflow.md.queue.QueueKeeper; +import org.opendaylight.openflowplugin.api.openflow.md.queue.QueueKeeper.QueueType; +import org.opendaylight.openflowplugin.api.openflow.md.queue.QueueProcessor; +import org.opendaylight.openflowplugin.api.openflow.md.queue.WaterMarkListener; +import org.opendaylight.openflowplugin.api.openflow.md.queue.WaterMarkListenerImpl; import org.opendaylight.openflowplugin.openflow.md.core.session.OFSessionUtil; import org.opendaylight.openflowplugin.openflow.md.core.session.PortFeaturesUtil; -import org.opendaylight.openflowplugin.openflow.md.core.session.SessionContext; -import org.opendaylight.openflowplugin.openflow.md.core.session.SessionManager; -import org.opendaylight.openflowplugin.openflow.md.queue.QueueKeeper; +import org.opendaylight.openflowplugin.openflow.md.queue.QueueKeeperFactory; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartRequestFlags; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInputBuilder; @@ -59,14 +70,22 @@ import com.google.common.util.concurrent.Futures; * @author mirehak */ public class ConnectionConductorImpl implements OpenflowProtocolListener, - SystemNotificationsListener, ConnectionConductor, ConnectionReadyListener, HandshakeListener { + SystemNotificationsListener, ConnectionConductor, + ConnectionReadyListener, HandshakeListener, NotificationEnqueuer, + AutoCloseable { + + /** + * ingress queue limit + */ + private static final int INGRESS_QUEUE_MAX_SIZE = 200; protected static final Logger LOG = LoggerFactory .getLogger(ConnectionConductorImpl.class); - /* variable to make BitMap-based negotiation enabled / disabled. - * it will help while testing and isolating issues related to processing of - * BitMaps from switches. + /* + * variable to make BitMap-based negotiation enabled / disabled. it will + * help while testing and isolating issues related to processing of BitMaps + * from switches. */ private boolean isBitmapNegotiationEnable = true; protected ErrorHandler errorHandler; @@ -79,24 +98,40 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, protected SessionContext sessionContext; - private QueueKeeper queueKeeper; - private ExecutorService hsPool; + private QueueProcessor queueProcessor; + private QueueKeeper queue; + private ThreadPoolExecutor hsPool; private HandshakeManager handshakeManager; private boolean firstHelloProcessed; - + private PortFeaturesUtil portFeaturesUtils; + private int conductorId; + + private int ingressMaxQueueSize; + /** * @param connectionAdapter */ public ConnectionConductorImpl(ConnectionAdapter connectionAdapter) { + this(connectionAdapter, INGRESS_QUEUE_MAX_SIZE); + } + + /** + * @param connectionAdapter + * @param ingressMaxQueueSize + * ingress queue limit (blocking) + */ + public ConnectionConductorImpl(ConnectionAdapter connectionAdapter, + int ingressMaxQueueSize) { this.connectionAdapter = connectionAdapter; + this.ingressMaxQueueSize = ingressMaxQueueSize; conductorState = CONDUCTOR_STATE.HANDSHAKING; - hsPool = Executors.newFixedThreadPool(1); firstHelloProcessed = false; handshakeManager = new HandshakeManagerImpl(connectionAdapter, - ConnectionConductor.versionOrder.get(0), ConnectionConductor.versionOrder); + ConnectionConductor.versionOrder.get(0), + ConnectionConductor.versionOrder); handshakeManager.setUseVersionBitmap(isBitmapNegotiationEnable); handshakeManager.setHandshakeListener(this); portFeaturesUtils = PortFeaturesUtil.getInstance(); @@ -104,18 +139,30 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, @Override public void init() { + int handshakeThreadLimit = 1; + hsPool = new ThreadPoolLoggingExecutor(handshakeThreadLimit, + handshakeThreadLimit, 0L, TimeUnit.MILLISECONDS, + new LinkedBlockingQueue(), "OFHandshake-" + + conductorId); + connectionAdapter.setMessageListener(this); connectionAdapter.setSystemListener(this); connectionAdapter.setConnectionReadyListener(this); + WaterMarkListener waterMarkListener = new WaterMarkListenerImpl( + connectionAdapter); + queue = QueueKeeperFactory.createFairQueueKeeper(queueProcessor, + ingressMaxQueueSize, waterMarkListener); } @Override - public void setQueueKeeper(QueueKeeper queueKeeper) { - this.queueKeeper = queueKeeper; + public void setQueueProcessor( + QueueProcessor queueProcessor) { + this.queueProcessor = queueProcessor; } /** - * @param errorHandler the errorHandler to set + * @param errorHandler + * the errorHandler to set */ @Override public void setErrorHandler(ErrorHandler errorHandler) { @@ -128,7 +175,8 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, new Thread(new Runnable() { @Override public void run() { - LOG.debug("echo request received: " + echoRequestMessage.getXid()); + LOG.debug("echo request received: " + + echoRequestMessage.getXid()); EchoReplyInputBuilder builder = new EchoReplyInputBuilder(); builder.setVersion(echoRequestMessage.getVersion()); builder.setXid(echoRequestMessage.getXid()); @@ -141,31 +189,52 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, @Override public void onErrorMessage(ErrorMessage errorMessage) { - queueKeeper.push(errorMessage, this); + enqueueMessage(errorMessage); + } + + /** + * @param message + */ + private void enqueueMessage(OfHeader message) { + enqueueMessage(message, QueueType.DEFAULT); + } + + @Override + public void enqueueNotification(NotificationQueueWrapper notification) { + enqueueMessage(notification); + } + + /** + * @param message + * @param queueType + * enqueue type + */ + private void enqueueMessage(OfHeader message, QueueType queueType) { + queue.push(message, this, queueType); } @Override public void onExperimenterMessage(ExperimenterMessage experimenterMessage) { - queueKeeper.push(experimenterMessage, this); + enqueueMessage(experimenterMessage); } @Override public void onFlowRemovedMessage(FlowRemovedMessage message) { - queueKeeper.push(message, this); + enqueueMessage(message); } - /** - * version negotiation happened as per following steps: - * 1. If HelloMessage version field has same version, continue connection processing. - * If HelloMessage version is lower than supported versions, just disconnect. + * version negotiation happened as per following steps: 1. If HelloMessage + * version field has same version, continue connection processing. If + * HelloMessage version is lower than supported versions, just disconnect. * 2. If HelloMessage contains bitmap and common version found in bitmap - * then continue connection processing. if no common version found, just disconnect. - * 3. If HelloMessage version is not supported, send HelloMessage with lower supported version. - * 4. If Hello message received again with not supported version, just disconnect. + * then continue connection processing. if no common version found, just + * disconnect. 3. If HelloMessage version is not supported, send + * HelloMessage with lower supported version. 4. If Hello message received + * again with not supported version, just disconnect. */ @Override - public synchronized void onHelloMessage(final HelloMessage hello) { + public void onHelloMessage(final HelloMessage hello) { LOG.debug("processing HELLO.xid: {}", hello.getXid()); firstHelloProcessed = true; checkState(CONDUCTOR_STATE.HANDSHAKING); @@ -192,20 +261,20 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, @Override public void onMultipartReplyMessage(MultipartReplyMessage message) { - queueKeeper.push(message, this); + enqueueMessage(message); } @Override public void onPacketInMessage(PacketInMessage message) { - queueKeeper.push(message, this, QueueKeeper.QueueType.UNORDERED); + enqueueMessage(message, QueueKeeper.QueueType.UNORDERED); } @Override public void onPortStatusMessage(PortStatusMessage message) { processPortStatusMsg(message); - queueKeeper.push(message, this); + enqueueMessage(message); } - + protected void processPortStatusMsg(PortStatus msg) { if (msg.getReason().getIntValue() == 2) { updatePort(msg); @@ -215,22 +284,25 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, deletePort(msg); } } - + protected void updatePort(PortStatus msg) { - Long portNumber = msg.getPortNo(); + Long portNumber = msg.getPortNo(); Boolean portBandwidth = portFeaturesUtils.getPortBandwidth(msg); - - if(portBandwidth == null) { - LOG.debug("can't get bandwidth info from port: {}, aborting port update", msg.toString()); + + if (portBandwidth == null) { + LOG.debug( + "can't get bandwidth info from port: {}, aborting port update", + msg.toString()); } else { this.getSessionContext().getPhysicalPorts().put(portNumber, msg); - this.getSessionContext().getPortsBandwidth().put(portNumber, portBandwidth); - } + this.getSessionContext().getPortsBandwidth() + .put(portNumber, portBandwidth); + } } - + protected void deletePort(PortGrouping port) { Long portNumber = port.getPortNo(); - + this.getSessionContext().getPhysicalPorts().remove(portNumber); this.getSessionContext().getPortsBandwidth().remove(portNumber); } @@ -241,12 +313,17 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, @Override public void run() { if (!CONDUCTOR_STATE.WORKING.equals(getConductorState())) { - // idle state in any other conductorState than WORKING means real - // problem and wont be handled by echoReply, but disconnection + // idle state in any other conductorState than WORKING means + // real + // problem and wont be handled by echoReply, but + // disconnection disconnect(); - OFSessionUtil.getSessionManager().invalidateOnDisconnect(ConnectionConductorImpl.this); + OFSessionUtil.getSessionManager().invalidateOnDisconnect( + ConnectionConductorImpl.this); } else { - LOG.debug("first idle state occured, sessionCtx={}|auxId={}", sessionContext, auxiliaryKey); + LOG.debug( + "first idle state occured, sessionCtx={}|auxId={}", + sessionContext, auxiliaryKey); EchoInputBuilder builder = new EchoInputBuilder(); builder.setVersion(getVersion()); builder.setXid(getSessionContext().getNextXid()); @@ -255,27 +332,30 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, .echo(builder.build()); try { - RpcResult echoReplyValue = echoReplyFuture.get(getMaxTimeout(), - getMaxTimeoutUnit()); + RpcResult echoReplyValue = echoReplyFuture + .get(getMaxTimeout(), getMaxTimeoutUnit()); if (echoReplyValue.isSuccessful()) { setConductorState(CONDUCTOR_STATE.WORKING); } else { - for (RpcError replyError : echoReplyValue.getErrors()) { + for (RpcError replyError : echoReplyValue + .getErrors()) { Throwable cause = replyError.getCause(); LOG.error( "while receiving echoReply in TIMEOUTING state: " + cause.getMessage(), cause); } - //switch issue occurred + // switch issue occurred throw new Exception("switch issue occurred"); } } catch (Exception e) { LOG.error("while waiting for echoReply in TIMEOUTING state: " + e.getMessage()); errorHandler.handleException(e, sessionContext); - //switch is not responding + // switch is not responding disconnect(); - OFSessionUtil.getSessionManager().invalidateOnDisconnect(ConnectionConductorImpl.this); + OFSessionUtil.getSessionManager() + .invalidateOnDisconnect( + ConnectionConductorImpl.this); } } } @@ -298,7 +378,7 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, } /** - * @param handshaking + * @param expectedState */ protected void checkState(CONDUCTOR_STATE expectedState) { if (!conductorState.equals(expectedState)) { @@ -311,6 +391,7 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, public void onDisconnectEvent(DisconnectEvent arg0) { SessionManager sessionManager = OFSessionUtil.getSessionManager(); sessionManager.invalidateOnDisconnect(this); + close(); } @Override @@ -320,7 +401,8 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, @Override public Future disconnect() { - LOG.trace("disconnecting: sessionCtx={}|auxId={}", sessionContext, auxiliaryKey); + LOG.trace("disconnecting: sessionCtx={}|auxId={}", sessionContext, + auxiliaryKey); Future result = null; if (connectionAdapter.isAlive()) { @@ -329,7 +411,7 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, LOG.debug("connection already disconnected"); result = Futures.immediateFuture(true); } - + close(); return result; } @@ -359,9 +441,9 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, } @Override - public synchronized void onConnectionReady() { + public void onConnectionReady() { LOG.debug("connection is ready-to-use"); - if (! firstHelloProcessed) { + if (!firstHelloProcessed) { HandshakeStepWrapper handshakeStepWrapper = new HandshakeStepWrapper( null, handshakeManager, connectionAdapter); hsPool.execute(handshakeStepWrapper); @@ -375,45 +457,60 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, public void onHandshakeSuccessfull(GetFeaturesOutput featureOutput, Short negotiatedVersion) { postHandshakeBasic(featureOutput, negotiatedVersion); - + // post-handshake actions - if(version == OFConstants.OFP_VERSION_1_3){ + if (version == OFConstants.OFP_VERSION_1_3) { + requestPorts(); requestGroupFeatures(); requestMeterFeatures(); - } else if (version == OFConstants.OFP_VERSION_1_0) { - // Because the GetFeaturesOutput contains information about the port - // in OF1.0 (that we would otherwise get from the PortDesc) we have to pass - // it up for parsing to convert into a NodeConnectorUpdate - queueKeeper.push(featureOutput, this); } - + requestDesc(); - requestPorts(); + } + + @Override + public void onHandshakeFailure() { + LOG.info("OF handshake failed, doing cleanup."); + close(); } /** * used by tests + * * @param featureOutput * @param negotiatedVersion */ protected void postHandshakeBasic(GetFeaturesOutput featureOutput, Short negotiatedVersion) { version = negotiatedVersion; - conductorState = CONDUCTOR_STATE.WORKING; + if (version == OFConstants.OFP_VERSION_1_0) { + // Because the GetFeaturesOutput contains information about the port + // in OF1.0 (that we would otherwise get from the PortDesc) we have + // to pass + // it up for parsing to convert into a NodeConnectorUpdate + // + // BUG-1988 - this must be the first item in queue in order not to + // get behind link-up message + enqueueMessage(featureOutput); + } + OFSessionUtil.registerSession(this, featureOutput, negotiatedVersion); - hsPool.shutdownNow(); + hsPool.shutdown(); + hsPool.purge(); + conductorState = CONDUCTOR_STATE.WORKING; + QueueKeeperFactory.plugQueue(queueProcessor, queue); } /* - * Send an OFPMP_DESC request message to the switch + * Send an OFPMP_DESC request message to the switch */ - private void requestDesc() { MultipartRequestInputBuilder builder = new MultipartRequestInputBuilder(); builder.setType(MultipartType.OFPMPDESC); builder.setVersion(getVersion()); builder.setFlags(new MultipartRequestFlags(false)); - builder.setMultipartRequestBody(new MultipartRequestDescCaseBuilder().build()); + builder.setMultipartRequestBody(new MultipartRequestDescCaseBuilder() + .build()); builder.setXid(getSessionContext().getNextXid()); getConnectionAdapter().multipartRequest(builder.build()); } @@ -423,45 +520,49 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, builder.setType(MultipartType.OFPMPPORTDESC); builder.setVersion(getVersion()); builder.setFlags(new MultipartRequestFlags(false)); - builder.setMultipartRequestBody(new MultipartRequestPortDescCaseBuilder().build()); + builder.setMultipartRequestBody(new MultipartRequestPortDescCaseBuilder() + .build()); builder.setXid(getSessionContext().getNextXid()); getConnectionAdapter().multipartRequest(builder.build()); } - private void requestGroupFeatures(){ + + private void requestGroupFeatures() { MultipartRequestInputBuilder mprInput = new MultipartRequestInputBuilder(); mprInput.setType(MultipartType.OFPMPGROUPFEATURES); mprInput.setVersion(getVersion()); mprInput.setFlags(new MultipartRequestFlags(false)); mprInput.setXid(getSessionContext().getNextXid()); - MultipartRequestGroupFeaturesCaseBuilder mprGroupFeaturesBuild = - new MultipartRequestGroupFeaturesCaseBuilder(); + MultipartRequestGroupFeaturesCaseBuilder mprGroupFeaturesBuild = new MultipartRequestGroupFeaturesCaseBuilder(); mprInput.setMultipartRequestBody(mprGroupFeaturesBuild.build()); - LOG.debug("Send group features statistics request :{}",mprGroupFeaturesBuild); + LOG.debug("Send group features statistics request :{}", + mprGroupFeaturesBuild); getConnectionAdapter().multipartRequest(mprInput.build()); - + } - private void requestMeterFeatures(){ + + private void requestMeterFeatures() { MultipartRequestInputBuilder mprInput = new MultipartRequestInputBuilder(); mprInput.setType(MultipartType.OFPMPMETERFEATURES); mprInput.setVersion(getVersion()); mprInput.setFlags(new MultipartRequestFlags(false)); mprInput.setXid(getSessionContext().getNextXid()); - MultipartRequestMeterFeaturesCaseBuilder mprMeterFeaturesBuild = - new MultipartRequestMeterFeaturesCaseBuilder(); + MultipartRequestMeterFeaturesCaseBuilder mprMeterFeaturesBuild = new MultipartRequestMeterFeaturesCaseBuilder(); mprInput.setMultipartRequestBody(mprMeterFeaturesBuild.build()); - LOG.debug("Send meter features statistics request :{}",mprMeterFeaturesBuild); + LOG.debug("Send meter features statistics request :{}", + mprMeterFeaturesBuild); getConnectionAdapter().multipartRequest(mprInput.build()); - + } + /** - * @param isBitmapNegotiationEnable the isBitmapNegotiationEnable to set + * @param isBitmapNegotiationEnable + * the isBitmapNegotiationEnable to set */ - public void setBitmapNegotiationEnable( - boolean isBitmapNegotiationEnable) { + public void setBitmapNegotiationEnable(boolean isBitmapNegotiationEnable) { this.isBitmapNegotiationEnable = isBitmapNegotiationEnable; } @@ -469,4 +570,27 @@ public class ConnectionConductorImpl implements OpenflowProtocolListener, hsPool.shutdownNow(); LOG.debug("pool is terminated: {}", hsPool.isTerminated()); } + + protected void shutdownPoolPolitely() { + hsPool.shutdown(); + try { + hsPool.awaitTermination(1, TimeUnit.SECONDS); + } catch (InterruptedException e) { + LOG.info("Error while awaiting termination on pool. Will use shutdownNow method."); + shutdownPool(); + } + hsPool.purge(); + LOG.debug("pool is terminated: {}", hsPool.isTerminated()); + } + + @Override + public void setId(int conductorId) { + this.conductorId = conductorId; + } + + @Override + public void close() { + shutdownPoolPolitely(); + conductorState = CONDUCTOR_STATE.RIP; + } }