X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=pcep%2Ftopology%2Ftopology-provider%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fbgpcep%2Fpcep%2Ftopology%2Fprovider%2FAbstractTopologySessionListener.java;h=f16b79d331005f1ad36ef67bc6c7782abd57fc20;hb=2fa6350d8a698badf8ccb81a34f2a2484b8976b7;hp=ff0ffc9255bf7dc9ce805708f880b55873cc64bd;hpb=1717bdf2c27f1a442c6589c624b7982e188cbe91;p=bgpcep.git diff --git a/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java b/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java index ff0ffc9255..f16b79d331 100644 --- a/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java +++ b/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java @@ -9,14 +9,13 @@ package org.opendaylight.bgpcep.pcep.topology.provider; import static java.util.Objects.requireNonNull; -import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; +import com.google.common.collect.Maps; import com.google.common.util.concurrent.FluentFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; -import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.netty.util.concurrent.FutureListener; import java.net.InetAddress; import java.util.ArrayList; @@ -29,11 +28,12 @@ import java.util.Objects; import java.util.Optional; import java.util.Timer; import java.util.TimerTask; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Stream; import org.checkerframework.checker.lock.qual.GuardedBy; import org.checkerframework.checker.lock.qual.Holding; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.bgpcep.pcep.topology.provider.session.stats.SessionStateImpl; import org.opendaylight.bgpcep.pcep.topology.provider.session.stats.TopologySessionStats; import org.opendaylight.mdsal.binding.api.WriteTransaction; @@ -51,6 +51,7 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.typ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.MessageHeader; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.Object; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.ProtocolVersion; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.open.object.open.Tlvs; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.LspId; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1Builder; @@ -64,6 +65,7 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLspBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLspKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.reported.lsp.Path; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.reported.lsp.PathKey; import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node; import org.opendaylight.yangtools.yang.binding.DataObject; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; @@ -82,6 +84,9 @@ import org.slf4j.LoggerFactory; * @param identifier type for LSPs */ public abstract class AbstractTopologySessionListener implements TopologySessionListener, TopologySessionStats { + private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class); + + static final String MISSING_XML_TAG = "Mandatory XML tags are missing."; static final MessageHeader MESSAGE_HEADER = new MessageHeader() { private final ProtocolVersion version = new ProtocolVersion(Uint8.ONE); @@ -92,19 +97,19 @@ public abstract class AbstractTopologySessionListener implements TopologyS @Override public ProtocolVersion getVersion() { - return this.version; + return version; } }; - static final String MISSING_XML_TAG = "Mandatory XML tags are missing."; - private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class); + @GuardedBy("this") final Map lsps = new HashMap<>(); @GuardedBy("this") - final SessionStateImpl listenerState; + SessionStateImpl listenerState; + @GuardedBy("this") private final Map requests = new HashMap<>(); @GuardedBy("this") - private final Map lspData = new HashMap<>(); + private final Map lspData = new ConcurrentHashMap<>(); private final ServerSessionManager serverSessionManager; private InstanceIdentifier pccIdentifier; @GuardedBy("this") @@ -119,66 +124,78 @@ public abstract class AbstractTopologySessionListener implements TopologyS AbstractTopologySessionListener(final ServerSessionManager serverSessionManager) { this.serverSessionManager = requireNonNull(serverSessionManager); - this.listenerState = new SessionStateImpl(this); } @Override - public final synchronized void onSessionUp(final PCEPSession psession) { - /* - * The session went up. Look up the router in Inventory model, - * create it if it is not there (marking that fact for later - * deletion), and mark it as synchronizing. Also create it in - * the topology model, with empty LSP list. - */ - final InetAddress peerAddress = psession.getRemoteAddress(); - - this.syncOptimization = new SyncOptimization(psession); - - final TopologyNodeState state = this.serverSessionManager.takeNodeState(peerAddress, - this, isLspDbRetreived()); - - // takeNodeState(..) may fail when the server session manager is being restarted due to configuration change - if (state == null) { - LOG.error("Unable to fetch topology node state for PCEP session. Closing session {}", psession); - psession.close(TerminationReason.UNKNOWN); - this.onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); - return; - } + public final void onSessionUp(final PCEPSession psession) { + synchronized (this.serverSessionManager) { + synchronized (this) { + /* + * The session went up. Look up the router in Inventory model, create it if it + * is not there (marking that fact for later deletion), and mark it as + * synchronizing. Also create it in the topology model, with empty LSP list. + */ + final InetAddress peerAddress = psession.getRemoteAddress(); + + this.syncOptimization = new SyncOptimization(psession); + final boolean haveLspDbVersion = this.syncOptimization.isDbVersionPresent(); + + final TopologyNodeState state = + this.serverSessionManager.takeNodeState(peerAddress, this, haveLspDbVersion); + + // takeNodeState(..) may fail when the server session manager is being restarted + // due to configuration change + if (state == null) { + LOG.error("Unable to fetch topology node state for PCEP session. Closing session {}", psession); + psession.close(TerminationReason.UNKNOWN); + this.onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); + return; + } - if (this.session != null || this.nodeState != null) { - LOG.error("PCEP session is already up with {}. Closing session {}", psession.getRemoteAddress(), psession); - psession.close(TerminationReason.UNKNOWN); - this.onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); - return; - } - this.session = psession; - this.nodeState = state; - - LOG.trace("Peer {} resolved to topology node {}", peerAddress, state.getNodeId()); - - // Our augmentation in the topology node - final PathComputationClientBuilder pccBuilder = new PathComputationClientBuilder(); - - onSessionUp(psession, pccBuilder); - this.synced.set(isSynchronized()); - - pccBuilder.setIpAddress(IetfInetUtil.INSTANCE.ipAddressNoZoneFor(peerAddress)); - final InstanceIdentifier topologyAugment = state.getNodeId().augmentation(Node1.class); - this.pccIdentifier = topologyAugment.child(PathComputationClient.class); - final Node initialNodeState = state.getInitialNodeState(); - final boolean isNodePresent = isLspDbRetreived() && initialNodeState != null; - if (isNodePresent) { - loadLspData(initialNodeState, this.lspData, this.lsps, isIncrementalSynchro()); - pccBuilder.setReportedLsp(initialNodeState.augmentation(Node1.class) - .getPathComputationClient().getReportedLsp()); + if (this.session != null || this.nodeState != null) { + LOG.error("PCEP session is already up with {}. Closing session {}", peerAddress, psession); + psession.close(TerminationReason.UNKNOWN); + this.onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); + return; + } + this.session = psession; + this.nodeState = state; + + LOG.trace("Peer {} resolved to topology node {}", peerAddress, state.getNodeId()); + + // Our augmentation in the topology node + final PathComputationClientBuilder pccBuilder = new PathComputationClientBuilder() + .setIpAddress(IetfInetUtil.INSTANCE.ipAddressNoZoneFor(peerAddress)); + + // Let subclass fill the details + onSessionUp(pccBuilder, peerAddress, psession.getRemoteTlvs()); + + this.synced.set(isSynchronized()); + + final InstanceIdentifier topologyAugment = state.getNodeId().augmentation(Node1.class); + this.pccIdentifier = topologyAugment.child(PathComputationClient.class); + + if (haveLspDbVersion) { + final Node initialNodeState = state.getInitialNodeState(); + if (initialNodeState != null) { + loadLspData(initialNodeState, this.lspData, this.lsps, isIncrementalSynchro()); + pccBuilder.setReportedLsp( + initialNodeState.augmentation(Node1.class).getPathComputationClient().getReportedLsp()); + } + } + state.storeNode(topologyAugment, + new Node1Builder().setPathComputationClient(pccBuilder.build()).build(), psession); + + this.listenerState = new SessionStateImpl(this, psession); + this.serverSessionManager.bind(state.getNodeId(), this.listenerState); + LOG.info("Session with {} attached to topology node {}", peerAddress, state.getNodeId()); + } } - state.storeNode(topologyAugment, - new Node1Builder().setPathComputationClient(pccBuilder.build()).build(), this.session); - this.listenerState.init(psession); - this.serverSessionManager.bind(this.nodeState.getNodeId(), this.listenerState); - LOG.info("Session with {} attached to topology node {}", psession.getRemoteAddress(), state.getNodeId()); } + protected abstract void onSessionUp(PathComputationClientBuilder pccBuilder, InetAddress peerAddress, + @Nullable Tlvs remoteTlvs); + synchronized void updatePccState(final PccSyncState pccSyncState) { if (this.nodeState == null) { LOG.info("Server Session Manager is closed."); @@ -216,58 +233,63 @@ public abstract class AbstractTopologySessionListener implements TopologyS * Tear down the given PCEP session. It's OK to call this method even after the session * is already down. It always clear up the current session status. */ - @Holding("this") @SuppressWarnings("checkstyle:IllegalCatch") - private synchronized void tearDown(final PCEPSession psession) { - + private void tearDown(final PCEPSession psession) { requireNonNull(psession); - this.serverSessionManager.releaseNodeState(this.nodeState, psession, isLspDbPersisted()); - clearNodeState(); - - try { - if (this.session != null) { - this.session.close(); - } - psession.close(); - } catch (final Exception e) { - LOG.error("Session {} cannot be closed.", psession, e); - } - this.session = null; - this.syncOptimization = null; - - // Clear all requests we know about - for (final Entry e : this.requests.entrySet()) { - final PCEPRequest r = e.getValue(); - switch (r.getState()) { - case DONE: - // Done is done, nothing to do - LOG.trace("Request {} was done when session went down.", e.getKey()); - break; - case UNACKED: - // Peer has not acked: results in failure - LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey()); - r.done(OperationResults.NOACK); - break; - case UNSENT: - // Peer has not been sent to the peer: results in cancellation - LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey()); - r.done(OperationResults.UNSENT); - break; - default: - break; + synchronized (this.serverSessionManager) { + synchronized (this) { + this.serverSessionManager.releaseNodeState(this.nodeState, psession, isLspDbPersisted()); + clearNodeState(); + + try { + if (this.session != null) { + this.session.close(); + } + psession.close(); + } catch (final Exception e) { + LOG.error("Session {} cannot be closed.", psession, e); + } + this.session = null; + this.listenerState = null; + this.syncOptimization = null; + + // Clear all requests we know about + for (final Entry e : this.requests.entrySet()) { + final PCEPRequest r = e.getValue(); + switch (r.getState()) { + case DONE: + // Done is done, nothing to do + LOG.trace("Request {} was done when session went down.", e.getKey()); + break; + case UNACKED: + // Peer has not acked: results in failure + LOG.info("Request {} was incomplete when session went down, failing the instruction", + e.getKey()); + r.done(OperationResults.NOACK); + break; + case UNSENT: + // Peer has not been sent to the peer: results in cancellation + LOG.debug("Request {} was not sent when session went down, cancelling the instruction", + e.getKey()); + r.done(OperationResults.UNSENT); + break; + default: + break; + } + } + this.requests.clear(); } } - this.requests.clear(); } @Override - public final synchronized void onSessionDown(final PCEPSession psession, final Exception exception) { + public final void onSessionDown(final PCEPSession psession, final Exception exception) { LOG.warn("Session {} went down unexpectedly", psession, exception); tearDown(psession); } @Override - public final synchronized void onSessionTerminated(final PCEPSession psession, final PCEPTerminationReason reason) { + public final void onSessionTerminated(final PCEPSession psession, final PCEPTerminationReason reason) { LOG.info("Session {} terminated by peer with reason {}", psession, reason); tearDown(psession); } @@ -315,25 +337,40 @@ public abstract class AbstractTopologySessionListener implements TopologyS }, MoreExecutors.directExecutor()); } + /** + * Perform revision-specific message processing when a message arrives. + * + * @param ctx Message processing context + * @param message Protocol message + * @return True if the message type is not handle. + */ + protected abstract boolean onMessage(MessageContext ctx, Message message); + @Override - public synchronized void close() { - clearNodeState(); - if (this.session != null) { - LOG.info("Closing session {}", session); - this.session.close(TerminationReason.UNKNOWN); + public void close() { + synchronized (this.serverSessionManager) { + synchronized (this) { + clearNodeState(); + if (this.session != null) { + LOG.info("Closing session {}", session); + this.session.close(TerminationReason.UNKNOWN); + } + } } } - private synchronized void clearNodeState() { + @Holding({"this.serverSessionManager", "this"}) + private void clearNodeState() { if (this.nodeState != null) { this.serverSessionManager.unbind(this.nodeState.getNodeId()); + LOG.debug("Clear Node state: {}", this.nodeState.getNodeId()); this.nodeState = null; } } final synchronized PCEPRequest removeRequest(final S id) { final PCEPRequest ret = this.requests.remove(id); - if (ret != null) { + if (ret != null && this.listenerState != null) { this.listenerState.processRequestStats(ret.getElapsedMillis()); } LOG.trace("Removed request {} object {}", id, ret); @@ -342,7 +379,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS final synchronized ListenableFuture sendMessage(final Message message, final S requestId, final Metadata metadata) { - final io.netty.util.concurrent.Future f = this.session.sendMessage(message); + final var sendFuture = this.session.sendMessage(message); this.listenerState.updateStatefulSentMsg(message); final PCEPRequest req = new PCEPRequest(metadata); this.requests.put(requestId, req); @@ -352,7 +389,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS setupTimeoutHandler(requestId, req, rpcTimeout); } - f.addListener((FutureListener) future -> { + sendFuture.addListener((FutureListener) future -> { if (!future.isSuccess()) { synchronized (AbstractTopologySessionListener.this) { AbstractTopologySessionListener.this.requests.remove(requestId); @@ -414,7 +451,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS final ReportedLsp previous = this.lspData.get(name); // if no previous report about the lsp exist, just proceed if (previous != null) { - final List updatedPaths = makeBeforeBreak(rlb, previous, name, remove); + final Map updatedPaths = makeBeforeBreak(rlb, previous, name, remove); // if all paths or the last path were deleted, delete whole tunnel if (updatedPaths.isEmpty()) { LOG.debug("All paths were removed, removing LSP with {}.", id); @@ -440,12 +477,11 @@ public abstract class AbstractTopologySessionListener implements TopologyS this.lspData.put(name, rl); } - private static List makeBeforeBreak(final ReportedLspBuilder rlb, final ReportedLsp previous, + private static Map makeBeforeBreak(final ReportedLspBuilder rlb, final ReportedLsp previous, final String name, final boolean remove) { // just one path should be reported final Path path = Iterables.getOnlyElement(rlb.getPath().values()); - final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev150820.LspId reportedLspId = - path.getLspId(); + final var reportedLspId = path.getLspId(); final List updatedPaths; //lspId = 0 and remove = false -> tunnel is down, still exists but no path is signaled //remove existing tunnel's paths now, as explicit path remove will not come @@ -485,7 +521,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS } } LOG.debug("Setting new paths {} to lsp {}", updatedPaths, name); - return updatedPaths; + return Maps.uniqueIndex(updatedPaths, Path::key); } /** @@ -529,22 +565,9 @@ public abstract class AbstractTopologySessionListener implements TopologyS this.lspData.remove(name); } - @SuppressWarnings("checkstyle:OverloadMethodsDeclarationOrder") - protected abstract void onSessionUp(PCEPSession session, PathComputationClientBuilder pccBuilder); - - /** - * Perform revision-specific message processing when a message arrives. - * - * @param ctx Message processing context - * @param message Protocol message - * @return True if the message type is not handle. - */ - @SuppressWarnings("checkstyle:OverloadMethodsDeclarationOrder") - protected abstract boolean onMessage(MessageContext ctx, Message message); - + @Holding("this") final String lookupLspName(final L id) { - requireNonNull(id, "ID parameter null."); - return this.lsps.get(id); + return this.lsps.get(requireNonNull(id, "ID parameter null.")); } /** @@ -571,10 +594,6 @@ public abstract class AbstractTopologySessionListener implements TopologyS return this.syncOptimization != null && this.syncOptimization.isSyncAvoidanceEnabled(); } - final boolean isLspDbRetreived() { - return this.syncOptimization != null && this.syncOptimization.isDbVersionPresent(); - } - /** * Is Incremental synchronization if LSP-DB-VERSION are included, * LSP-DB-VERSION TLV values doesnt match, and LSP-SYNC-CAPABILITY is enabled. @@ -598,18 +617,13 @@ public abstract class AbstractTopologySessionListener implements TopologyS @Override public int getDelegatedLspsCount() { - final Stream stream; - synchronized (this) { - stream = ImmutableList.copyOf(this.lspData.values()).stream(); - } - - return Math.toIntExact(stream + return Math.toIntExact(this.lspData.values().stream() .map(ReportedLsp::getPath).filter(pathList -> pathList != null && !pathList.isEmpty()) // pick the first path, as delegate status should be same in each path .map(pathList -> pathList.values().iterator().next()) .map(path -> path.augmentation(Path1.class)).filter(Objects::nonNull) .map(LspObject::getLsp).filter(Objects::nonNull) - .filter(Lsp::isDelegate) + .filter(Lsp::getDelegate) .count()); } @@ -633,13 +647,11 @@ public abstract class AbstractTopologySessionListener implements TopologyS } void resolveRequest(final PCEPRequest req) { - this.requests.add(req); + requests.add(req); } - @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", - justification = "https://github.com/spotbugs/spotbugs/issues/811") private void notifyRequests() { - for (final PCEPRequest r : this.requests) { + for (final PCEPRequest r : requests) { r.done(OperationResults.SUCCESS); } }