| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| |
| package org.apache.geode.internal.cache.locks; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.HashSet; |
| import java.util.Set; |
| import java.util.concurrent.RejectedExecutionException; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.MessageWithReply; |
| import org.apache.geode.distributed.internal.PooledDistributionMessage; |
| import org.apache.geode.distributed.internal.ReplyException; |
| import org.apache.geode.distributed.internal.ReplyMessage; |
| import org.apache.geode.distributed.internal.ReplyProcessor21; |
| import org.apache.geode.distributed.internal.locks.DLockGrantor; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.cache.TXCommitMessage; |
| import org.apache.geode.internal.logging.LogService; |
| |
| /** |
| * Sends <code>TXOriginatorRecoveryMessage</code> to all participants of a given transaction when |
| * the originator departs. The participants delay reply until the commit has finished. Once all |
| * replies have come in, the transaction lock (<code>TXLockId</code>) will be released. |
| * |
| */ |
| public class TXOriginatorRecoveryProcessor extends ReplyProcessor21 { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| static void sendMessage(Set members, InternalDistributedMember originator, TXLockId txLockId, |
| DLockGrantor grantor, DistributionManager dm) { |
| TXOriginatorRecoveryProcessor processor = new TXOriginatorRecoveryProcessor(dm, members); |
| |
| TXOriginatorRecoveryMessage msg = new TXOriginatorRecoveryMessage(); |
| msg.processorId = processor.getProcessorId(); |
| msg.txLockId = txLockId; |
| |
| // send msg to all members EXCEPT this member... |
| Set recipients = new HashSet(members); |
| recipients.remove(dm.getId()); |
| msg.setRecipients(members); |
| if (logger.isDebugEnabled()) { |
| logger.debug("Sending TXOriginatorRecoveryMessage: {}", msg); |
| } |
| dm.putOutgoing(msg); |
| |
| // process msg and reply directly if this VM is a participant... |
| if (members.contains(dm.getId())) { |
| if (msg.getSender() == null) |
| msg.setSender(dm.getId()); |
| msg.process((ClusterDistributionManager) dm); |
| } |
| |
| // keep waiting even if interrupted |
| // for() loop removed for bug 36983 - you can't loop on waitForReplies() |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| try { |
| processor.waitForRepliesUninterruptibly(); |
| } catch (ReplyException e) { |
| e.handleCause(); |
| } |
| |
| // release txLockId... |
| if (logger.isDebugEnabled()) { |
| logger.debug("TXOriginatorRecoveryProcessor releasing: {}", txLockId); |
| } |
| // dtls.release(txLockId); |
| try { |
| grantor.releaseLockBatch(txLockId, originator); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| // ------------------------------------------------------------------------- |
| // Constructors |
| // ------------------------------------------------------------------------- |
| |
| /** Creates a new instance of TXOriginatorRecoveryProcessor */ |
| private TXOriginatorRecoveryProcessor(DistributionManager dm, Set members) { |
| super(dm, members); |
| } |
| |
| @Override |
| protected boolean allowReplyFromSender() { |
| return true; |
| } |
| |
| /** |
| * IllegalStateException is an anticipated reply exception. Receiving multiple replies with this |
| * exception is normal. |
| */ |
| @Override |
| protected boolean logMultipleExceptions() { |
| return false; |
| } |
| |
| // ------------------------------------------------------------------------- |
| // TXOriginatorRecoveryMessage |
| // ------------------------------------------------------------------------- |
| public static class TXOriginatorRecoveryMessage extends PooledDistributionMessage |
| implements MessageWithReply { |
| |
| /** The transaction lock for which the originator orphaned */ |
| protected TXLockId txLockId; |
| |
| /** The reply processor to route replies to */ |
| protected int processorId; |
| |
| @Override |
| public int getProcessorId() { |
| return this.processorId; |
| } |
| |
| @Override |
| protected void process(final ClusterDistributionManager dm) { |
| final TXOriginatorRecoveryMessage msg = this; |
| |
| try { |
| dm.getWaitingThreadPool().execute(new Runnable() { |
| @Override |
| public void run() { |
| processTXOriginatorRecoveryMessage(dm, msg); |
| } |
| }); |
| } catch (RejectedExecutionException e) { |
| logger.debug("Rejected processing of <{}>", msg, e); |
| } |
| } |
| |
| protected void processTXOriginatorRecoveryMessage(final ClusterDistributionManager dm, |
| final TXOriginatorRecoveryMessage msg) { |
| |
| ReplyException replyException = null; |
| logger.info("[processTXOriginatorRecoveryMessage]"); |
| try { |
| // Wait for the transaction associated with this lockid to finish processing |
| TXCommitMessage.getTracker().waitToProcess(msg.txLockId, dm); |
| |
| // when the grantor receives reply it will release txLock... |
| /* |
| * TODO: implement waitToReleaseTXLockId here testTXOriginatorRecoveryProcessor in |
| * org.apache.geode.internal.cache.locks.TXLockServiceTest should be expanded upon also... |
| */ |
| } catch (RuntimeException t) { |
| logger.warn("[processTXOriginatorRecoveryMessage] throwable:", |
| t); |
| // if (replyException == null) (can only be null) |
| { |
| replyException = new ReplyException(t); |
| } |
| // else { |
| // log.warning(LocalizedStrings.TXOriginatorRecoveryProcessor_MORE_THAN_ONE_EXCEPTION_THROWN_IN__0, |
| // this, t); |
| // } |
| // } |
| // catch (VirtualMachineError err) { |
| // SystemFailure.initiateFailure(err); |
| // // If this ever returns, rethrow the error. We're poisoned |
| // // now, so don't let this thread continue. |
| // throw err; |
| // } |
| // catch (Throwable t) { |
| // // Whenever you catch Error or Throwable, you must also |
| // // catch VirtualMachineError (see above). However, there is |
| // // _still_ a possibility that you are dealing with a cascading |
| // // error condition, so you also need to check to see if the JVM |
| // // is still usable: |
| // SystemFailure.checkFailure(); |
| // if (replyException == null) { |
| // replyException = new ReplyException(t); |
| // } |
| // } |
| // catch (VirtualMachineError err) { |
| // SystemFailure.initiateFailure(err); |
| // // If this ever returns, rethrow the error. We're poisoned |
| // // now, so don't let this thread continue. |
| // throw err; |
| // } |
| // catch (Throwable t) { |
| // // Whenever you catch Error or Throwable, you must also |
| // // catch VirtualMachineError (see above). However, there is |
| // // _still_ a possibility that you are dealing with a cascading |
| // // error condition, so you also need to check to see if the JVM |
| // // is still usable: |
| // SystemFailure.checkFailure(); |
| // if (replyException == null) { |
| // replyException = new ReplyException(t); |
| // } |
| // } |
| } finally { |
| TXOriginatorRecoveryReplyMessage replyMsg = new TXOriginatorRecoveryReplyMessage(); |
| replyMsg.txLockId = txLockId; |
| replyMsg.setProcessorId(getProcessorId()); |
| replyMsg.setRecipient(getSender()); |
| replyMsg.setException(replyException); |
| |
| if (getSender().equals(dm.getId())) { |
| // process in-line in this VM |
| logger.info("[processTXOriginatorRecoveryMessage] locally process reply"); |
| replyMsg.setSender(dm.getId()); |
| replyMsg.dmProcess(dm); |
| } else { |
| logger.info("[processTXOriginatorRecoveryMessage] send reply"); |
| dm.putOutgoing(replyMsg); |
| } |
| } |
| } |
| |
| @Override |
| public int getDSFID() { |
| return TX_ORIGINATOR_RECOVERY_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| super.fromData(in); |
| this.txLockId = (TXLockId) DataSerializer.readObject(in); |
| this.processorId = in.readInt(); |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| super.toData(out); |
| DataSerializer.writeObject(this.txLockId, out); |
| out.writeInt(this.processorId); |
| } |
| |
| @Override |
| public String toString() { |
| StringBuffer buff = new StringBuffer(); |
| buff.append("TXOriginatorRecoveryMessage (txLockId='"); |
| buff.append(this.txLockId); |
| buff.append("'; processorId="); |
| buff.append(this.processorId); |
| buff.append(")"); |
| return buff.toString(); |
| } |
| } |
| |
| // ------------------------------------------------------------------------- |
| // TXOriginatorRecoveryReplyMessage |
| // ------------------------------------------------------------------------- |
| public static class TXOriginatorRecoveryReplyMessage extends ReplyMessage { |
| |
| /** The transaction lock for which the originator orphaned */ |
| protected TXLockId txLockId; // only for the toString |
| |
| @Override |
| public int getDSFID() { |
| return TX_ORIGINATOR_RECOVERY_REPLY_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| super.fromData(in); |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| super.toData(out); |
| } |
| |
| @Override |
| public String toString() { |
| return "TXOriginatorRecoveryReplyMessage (processorId=" + super.processorId + "; txLockId=" |
| + this.txLockId + "; sender=" + getSender() + ")"; |
| } |
| } |
| |
| } |
