import java.util.List;
import java.util.Queue;
import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.opendaylight.controller.config.util.xml.XmlElement;
private NetconfClientSession session;
private Future<?> initFuture;
- private SettableFuture<NetconfDeviceCapabilities> firstConnectionFuture;
+ private final SettableFuture<NetconfDeviceCapabilities> firstConnectionFuture;
+
+ // isSessionClosing indicates a close operation on the session is issued and
+ // tearDown will surely be called later to finish the close.
+ // Used to allow only one thread to enter tearDown and other threads should
+ // NOT enter it simultaneously and should end its close operation without
+ // calling tearDown to release the locks they hold to avoid deadlock.
+ private final AtomicBoolean isSessionClosing = new AtomicBoolean(false);
+
+ public Boolean isSessionClosing() {
+ return isSessionClosing.get();
+ }
public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
final UserPreferences NetconfSessionPreferences, final int rpcMessageLimit) {
LOG.trace("{}: Session advertised capabilities: {}", id,
netconfSessionPreferences);
- if(overrideNetconfCapabilities.isPresent()) {
- netconfSessionPreferences = overrideNetconfCapabilities.get().isOverride() ?
- netconfSessionPreferences.replaceModuleCaps(overrideNetconfCapabilities.get().getSessionPreferences()) :
- netconfSessionPreferences.addModuleCaps(overrideNetconfCapabilities.get().getSessionPreferences());
- LOG.debug(
- "{}: Session capabilities overridden, capabilities that will be used: {}",
- id, netconfSessionPreferences);
+ if (overrideNetconfCapabilities.isPresent()) {
+ final NetconfSessionPreferences sessionPreferences = overrideNetconfCapabilities
+ .get().getSessionPreferences();
+ netconfSessionPreferences = overrideNetconfCapabilities.get().moduleBasedCapsOverrided()
+ ? netconfSessionPreferences.replaceModuleCaps(sessionPreferences)
+ : netconfSessionPreferences.addModuleCaps(sessionPreferences);
+
+ netconfSessionPreferences = overrideNetconfCapabilities.get().nonModuleBasedCapsOverrided()
+ ? netconfSessionPreferences.replaceNonModuleCaps(sessionPreferences)
+ : netconfSessionPreferences.addNonModuleCaps(sessionPreferences);
+ LOG.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id,
+ netconfSessionPreferences);
}
-
remoteDevice.onRemoteSessionUp(netconfSessionPreferences, this);
if (!firstConnectionFuture.isDone()) {
firstConnectionFuture.set(netconfSessionPreferences.getNetconfDeviceCapabilities());
initFuture.addListener(new GenericFutureListener<Future<Object>>(){
@Override
- public void operationComplete(Future<Object> future) throws Exception {
+ public void operationComplete(final Future<Object> future) throws Exception {
if (!future.isSuccess() && !future.isCancelled()) {
LOG.debug("{}: Connection failed", id, future.cause());
NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause());
}
public void disconnect() {
- if(session != null) {
+ // If session is already in closing, no need to close it again
+ if(session != null && isSessionClosing.compareAndSet(false, true)) {
session.close();
}
}
- private void tearDown( String reason ) {
+ private void tearDown(final String reason) {
+ if (!isSessionClosing()) {
+ LOG.warn("It's curious that no one to close the session but tearDown is called!");
+ }
LOG.debug("Tearing down {}", reason);
- List<UncancellableFuture<RpcResult<NetconfMessage>>> futuresToCancel = Lists.newArrayList();
+ final List<UncancellableFuture<RpcResult<NetconfMessage>>> futuresToCancel = Lists.newArrayList();
sessionLock.lock();
try {
if( session != null ) {
// Notify pending request futures outside of the sessionLock to avoid unnecessarily
// blocking the caller.
- for( UncancellableFuture<RpcResult<NetconfMessage>> future: futuresToCancel ) {
+ for (final UncancellableFuture<RpcResult<NetconfMessage>> future : futuresToCancel) {
if( Strings.isNullOrEmpty( reason ) ) {
future.set( createSessionDownRpcResult() );
} else {
future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT, reason ) );
}
}
+
+ isSessionClosing.set(false);
}
private RpcResult<NetconfMessage> createSessionDownRpcResult() {
String.format( "The netconf session to %1$s is disconnected", id.getName() ) );
}
- private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message ) {
+ private RpcResult<NetconfMessage> createErrorRpcResult(final RpcError.ErrorType errorType, final String message) {
return RpcResultBuilder.<NetconfMessage>failed()
- .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build();
+ .withError(errorType, NetconfDocumentedException.ErrorTag.OPERATION_FAILED.getTagValue(), message).build();
}
@Override
public void onSessionDown(final NetconfClientSession session, final Exception e) {
- LOG.warn("{}: Session went down", id, e);
- tearDown( null );
+ // If session is already in closing, no need to call tearDown again.
+ if (isSessionClosing.compareAndSet(false, true)) {
+ LOG.warn("{}: Session went down", id, e);
+ tearDown( null );
+ }
}
@Override
public void onSessionTerminated(final NetconfClientSession session, final NetconfTerminationReason reason) {
+ // onSessionTerminated is called directly by disconnect, no need to compare and set isSessionClosing.
LOG.warn("{}: Session terminated {}", id, reason);
tearDown( reason.getErrorMessage() );
}
initFuture.cancel(false);
}
// Disconnect from device
- if(session != null) {
- session.close();
- // tear down not necessary, called indirectly by above close
- }
+ // tear down not necessary, called indirectly by the close in disconnect()
+ disconnect();
}
@Override
return Futures.immediateFuture( createSessionDownRpcResult() );
}
- final Request req = new Request( new UncancellableFuture<RpcResult<NetconfMessage>>(true),
- message );
+ final Request req = new Request(new UncancellableFuture<>(true), message);
requests.add(req);
session.sendMessage(req.request).addListener(new FutureListener<Void>() {
Code Review / netconf.git / blobdiff