import java.util.Iterator;
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;
protected final RemoteDeviceId id;
private final Lock sessionLock = new ReentrantLock();
- // TODO implement concurrent message limit
+ private final Semaphore semaphore;
+ private final int concurentRpcMsgs;
+
private final Queue<Request> requests = new ArrayDeque<>();
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) {
- this(id, remoteDevice, Optional.of(NetconfSessionPreferences));
+ final UserPreferences NetconfSessionPreferences, final int rpcMessageLimit) {
+ this(id, remoteDevice, Optional.of(NetconfSessionPreferences), rpcMessageLimit);
}
public NetconfDeviceCommunicator(final RemoteDeviceId id,
- final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice) {
- this(id, remoteDevice, Optional.<UserPreferences>absent());
+ final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
+ final int rpcMessageLimit) {
+ this(id, remoteDevice, Optional.<UserPreferences>absent(), rpcMessageLimit);
}
private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
- final Optional<UserPreferences> overrideNetconfCapabilities) {
+ final Optional<UserPreferences> overrideNetconfCapabilities, final int rpcMessageLimit) {
+ this.concurentRpcMsgs = rpcMessageLimit;
this.id = id;
this.remoteDevice = remoteDevice;
this.overrideNetconfCapabilities = overrideNetconfCapabilities;
this.firstConnectionFuture = SettableFuture.create();
+ this.semaphore = rpcMessageLimit > 0 ? new Semaphore(rpcMessageLimit) : null;
}
@Override
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
request = requests.peek();
if (request != null && request.future.isUncancellable()) {
requests.poll();
+ // we have just removed one request from the queue
+ // we can also release one permit
+ if(semaphore != null) {
+ semaphore.release();
+ }
} else {
request = null;
LOG.warn("{}: Ignoring unsolicited message {}", id,
@Override
public ListenableFuture<RpcResult<NetconfMessage>> sendRequest(final NetconfMessage message, final QName rpc) {
sessionLock.lock();
+
+ if (semaphore != null && !semaphore.tryAcquire()) {
+ LOG.warn("Limit of concurrent rpc messages was reached (limit :" +
+ concurentRpcMsgs + "). Rpc reply message is needed. Discarding request of Netconf device with id" + id.getName());
+ sessionLock.unlock();
+ return Futures.immediateFailedFuture(new NetconfDocumentedException("Limit of rpc messages was reached (Limit :" +
+ concurentRpcMsgs + ") waiting for emptying the queue of Netconf device with id" + id.getName()));
+ }
+
try {
- return sendRequestWithLock( message, rpc );
+ return sendRequestWithLock(message, rpc);
} finally {
sessionLock.unlock();
}
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>() {