- // Mainly checking for state violation here to perform a volatile read of "initialized" to
- // ensure updates to operationLimter et al are visible to this thread (ie we're doing
- // "piggy-back" synchronization here).
- Preconditions.checkState(initialized, "Tx was not propertly initialized.");
-
- // Create the TransactionContext from the response or failure. Store the new
- // TransactionContext locally until we've completed invoking the
- // TransactionOperations. This avoids thread timing issues which could cause
- // out-of-order TransactionOperations. Eg, on a modification operation, if the
- // TransactionContext is non-null, then we directly call the TransactionContext.
- // However, at the same time, the code may be executing the cached
- // TransactionOperations. So to avoid thus timing, we don't publish the
- // TransactionContext until after we've executed all cached TransactionOperations.
- TransactionContext localTransactionContext;
- if(failure != null) {
- LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier,
- failure.getMessage());
-
- localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
- } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
- localTransactionContext = createValidTransactionContext(
- CreateTransactionReply.fromSerializable(response));
- } else {
- IllegalArgumentException exception = new IllegalArgumentException(String.format(
- "Invalid reply type %s for CreateTransaction", response.getClass()));
-
- localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
- }
-
- executeTxOperatonsOnComplete(localTransactionContext);
- }
-
- private void executeTxOperatonsOnComplete(TransactionContext localTransactionContext) {
- while(true) {
- // Access to txOperationsOnComplete and transactionContext must be protected and atomic
- // (ie synchronized) with respect to #addTxOperationOnComplete to handle timing
- // issues and ensure no TransactionOperation is missed and that they are processed
- // in the order they occurred.
-
- // We'll make a local copy of the txOperationsOnComplete list to handle re-entrancy
- // in case a TransactionOperation results in another transaction operation being
- // queued (eg a put operation from a client read Future callback that is notified
- // synchronously).
- Collection<TransactionOperation> operationsBatch = null;
- synchronized(txOperationsOnComplete) {
- if(txOperationsOnComplete.isEmpty()) {
- // We're done invoking the TransactionOperations so we can now publish the
- // TransactionContext.
- transactionContext = localTransactionContext;
- break;
- }
-
- operationsBatch = new ArrayList<>(txOperationsOnComplete);
- txOperationsOnComplete.clear();
- }
-
- // Invoke TransactionOperations outside the sync block to avoid unnecessary blocking.
- // A slight down-side is that we need to re-acquire the lock below but this should
- // be negligible.
- for(TransactionOperation oper: operationsBatch) {
- oper.invoke(localTransactionContext);
- }
- }
- }
-
- private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
- String transactionPath = reply.getTransactionPath();
-
- LOG.debug("Tx {} Received {}", identifier, reply);
-
- ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
-
- if (transactionType == TransactionType.READ_ONLY) {
- // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
- // to close the remote Tx's when this instance is no longer in use and is garbage
- // collected.
-
- if(remoteTransactionActorsMB == null) {
- remoteTransactionActors = Lists.newArrayList();
- remoteTransactionActorsMB = new AtomicBoolean();