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;
+ // 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 volatile 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
}
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 ) {
+ 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();
sessionLock.lock();
future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT, reason ) );
}
}
+
+ isSessionClosing.set(false);
}
private RpcResult<NetconfMessage> createSessionDownRpcResult() {
@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() );
}
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();
}