import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
private final String transactionChainId;
private final SchemaContext schemaContext;
private boolean inReadyState;
+ private final Semaphore operationLimiter;
+ private final OperationCompleter operationCompleter;
public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
this(actorContext, transactionType, "");
phantomReferenceCache.put(cleanup, cleanup);
}
+ // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
+ this.operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
+ this.operationCompleter = new OperationCompleter(operationLimiter);
+
LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
}
LOG.debug("Tx {} read {}", identifier, path);
+ throttleOperation();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
return txFutureCallback.enqueueReadOperation(new ReadOperation<Optional<NormalizedNode<?, ?>>>() {
@Override
LOG.debug("Tx {} exists {}", identifier, path);
+ throttleOperation();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
return txFutureCallback.enqueueReadOperation(new ReadOperation<Boolean>() {
@Override
"Transaction is sealed - further modifications are not allowed");
}
+ private void throttleOperation() {
+ throttleOperation(1);
+ }
+
+ private void throttleOperation(int acquirePermits) {
+ try {
+ if(!operationLimiter.tryAcquire(acquirePermits, actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
+ LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
+ }
+ } catch (InterruptedException e) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Interrupted when trying to acquire operation permit for transaction " + getIdentifier().toString(), e);
+ } else {
+ LOG.warn("Interrupted when trying to acquire operation permit for transaction {}", getIdentifier());
+ }
+ }
+ }
+
+
@Override
public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
LOG.debug("Tx {} write {}", identifier, path);
+ throttleOperation();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
@Override
LOG.debug("Tx {} merge {}", identifier, path);
+ throttleOperation();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
@Override
LOG.debug("Tx {} delete {}", identifier, path);
+ throttleOperation();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
@Override
checkModificationState();
+ throttleOperation(txFutureCallbackMap.size());
+
inReadyState = true;
LOG.debug("Tx {} Readying {} transactions for commit", identifier,
LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier,
failure.getMessage());
- localTransactionContext = new NoOpTransactionContext(failure, identifier);
+ localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
} else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
localTransactionContext = createValidTransactionContext(
CreateTransactionReply.fromSerializable(response));
IllegalArgumentException exception = new IllegalArgumentException(String.format(
"Invalid reply type %s for CreateTransaction", response.getClass()));
- localTransactionContext = new NoOpTransactionContext(exception, identifier);
+ localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
}
for(TransactionOperation oper: txOperationsOnComplete) {
boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
return new TransactionContextImpl(transactionPath, transactionActor, identifier,
- actorContext, schemaContext, isTxActorLocal, reply.getVersion());
+ actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter);
}
}
private final Logger LOG = LoggerFactory.getLogger(TransactionContextImpl.class);
private final ActorContext actorContext;
- private final SchemaContext schemaContext;
private final String transactionPath;
private final ActorSelection actor;
private final boolean isTxActorLocal;
private final short remoteTransactionVersion;
+ private final OperationCompleter operationCompleter;
+
private TransactionContextImpl(String transactionPath, ActorSelection actor, TransactionIdentifier identifier,
ActorContext actorContext, SchemaContext schemaContext,
- boolean isTxActorLocal, short remoteTransactionVersion) {
+ boolean isTxActorLocal, short remoteTransactionVersion, OperationCompleter operationCompleter) {
super(identifier);
this.transactionPath = transactionPath;
this.actor = actor;
this.actorContext = actorContext;
- this.schemaContext = schemaContext;
this.isTxActorLocal = isTxActorLocal;
this.remoteTransactionVersion = remoteTransactionVersion;
+ this.operationCompleter = operationCompleter;
}
+ private Future<Object> completeOperation(Future<Object> operationFuture){
+ operationFuture.onComplete(this.operationCompleter, actorContext.getActorSystem().dispatcher());
+ return operationFuture;
+ }
+
+
private ActorSelection getActor() {
return actor;
}
private Future<Object> executeOperationAsync(SerializableMessage msg) {
- return actorContext.executeOperationAsync(getActor(), isTxActorLocal ? msg : msg.toSerializable());
+ return completeOperation(actorContext.executeOperationAsync(getActor(), isTxActorLocal ? msg : msg.toSerializable()));
}
private Future<Object> executeOperationAsync(VersionedSerializableMessage msg) {
- return actorContext.executeOperationAsync(getActor(), isTxActorLocal ? msg :
- msg.toSerializable(remoteTransactionVersion));
+ return completeOperation(actorContext.executeOperationAsync(getActor(), isTxActorLocal ? msg :
+ msg.toSerializable(remoteTransactionVersion)));
}
@Override
private final Logger LOG = LoggerFactory.getLogger(NoOpTransactionContext.class);
private final Throwable failure;
+ private final Semaphore operationLimiter;
- public NoOpTransactionContext(Throwable failure, TransactionIdentifier identifier){
+ public NoOpTransactionContext(Throwable failure, TransactionIdentifier identifier, Semaphore operationLimiter){
super(identifier);
this.failure = failure;
+ this.operationLimiter = operationLimiter;
}
@Override
@Override
public Future<ActorSelection> readyTransaction() {
LOG.debug("Tx {} readyTransaction called", identifier);
+ operationLimiter.release();
return akka.dispatch.Futures.failed(failure);
}
@Override
public void deleteData(YangInstanceIdentifier path) {
LOG.debug("Tx {} deleteData called path = {}", identifier, path);
+ operationLimiter.release();
}
@Override
public void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
LOG.debug("Tx {} mergeData called path = {}", identifier, path);
+ operationLimiter.release();
}
@Override
public void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
LOG.debug("Tx {} writeData called path = {}", identifier, path);
+ operationLimiter.release();
}
@Override
public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
YangInstanceIdentifier path) {
LOG.debug("Tx {} readData called path = {}", identifier, path);
+ operationLimiter.release();
return Futures.immediateFailedCheckedFuture(new ReadFailedException(
"Error reading data for path " + path, failure));
}
public CheckedFuture<Boolean, ReadFailedException> dataExists(
YangInstanceIdentifier path) {
LOG.debug("Tx {} dataExists called path = {}", identifier, path);
+ operationLimiter.release();
return Futures.immediateFailedCheckedFuture(new ReadFailedException(
"Error checking exists for path " + path, failure));
}
}
+
+ private static class OperationCompleter extends OnComplete<Object> {
+ private final Semaphore operationLimiter;
+ OperationCompleter(Semaphore operationLimiter){
+ this.operationLimiter = operationLimiter;
+ }
+
+ @Override
+ public void onComplete(Throwable throwable, Object o){
+ this.operationLimiter.release();
+ }
+ }
}
Code Review / controller.git / blobdiff