| /* |
| * 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; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.util.HashSet; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.annotations.Immutable; |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.execute.Function; |
| import org.apache.geode.cache.execute.FunctionException; |
| import org.apache.geode.cache.execute.FunctionService; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.DistributionMessage; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.OperationExecutors; |
| 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.membership.InternalDistributedMember; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.serialization.DeserializationContext; |
| import org.apache.geode.internal.serialization.SerializationContext; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| public class DistributedRegionFunctionStreamingMessage extends DistributionMessage |
| implements TransactionMessage { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| private Object result; |
| |
| transient int replyMsgNum = 0; |
| |
| transient boolean replyLastMsg; |
| |
| transient int numObjectsInChunk = 0; |
| |
| private Function functionObject; |
| |
| private String functionName; |
| |
| Object args; |
| |
| private String regionPath; |
| |
| private Set filter; |
| |
| private int processorId; |
| |
| private boolean isReExecute; |
| |
| private boolean isFnSerializationReqd; |
| |
| private int txUniqId = TXManagerImpl.NOTX; |
| |
| private InternalDistributedMember txMemberId = null; |
| |
| private static final short IS_REEXECUTE = UNRESERVED_FLAGS_START; |
| |
| /** default exception to ensure a false-positive response is never returned */ |
| @Immutable |
| static final ForceReattemptException UNHANDLED_EXCEPTION = |
| (ForceReattemptException) new ForceReattemptException("Unknown exception").fillInStackTrace(); |
| |
| public DistributedRegionFunctionStreamingMessage() {} |
| |
| public DistributedRegionFunctionStreamingMessage(final String regionPath, Function function, |
| int procId, final Set filter, Object args, boolean isReExecute, |
| boolean isFnSerializationReqd) { |
| this.functionObject = function; |
| this.processorId = procId; |
| this.args = args; |
| this.regionPath = regionPath; |
| this.filter = filter; |
| this.isReExecute = isReExecute; |
| this.isFnSerializationReqd = isFnSerializationReqd; |
| this.txUniqId = TXManagerImpl.getCurrentTXUniqueId(); |
| TXStateProxy txState = TXManagerImpl.getCurrentTXState(); |
| if (txState != null && txState.isMemberIdForwardingRequired()) { |
| this.txMemberId = txState.getOriginatingMember(); |
| } |
| } |
| |
| private TXStateProxy prepForTransaction(ClusterDistributionManager dm) |
| throws InterruptedException { |
| if (this.txUniqId == TXManagerImpl.NOTX) { |
| return null; |
| } else { |
| InternalCache cache = dm.getCache(); |
| if (cache == null) { |
| // ignore and return, we are shutting down! |
| return null; |
| } |
| TXManagerImpl mgr = cache.getTXMgr(); |
| return mgr.masqueradeAs(this); |
| } |
| } |
| |
| private void cleanupTransaction(TXStateProxy tx) { |
| if (this.txUniqId != TXManagerImpl.NOTX) { |
| InternalCache cache = GemFireCacheImpl.getInstance(); |
| if (cache == null) { |
| // ignore and return, we are shutting down! |
| return; |
| } |
| TXManagerImpl mgr = cache.getTXMgr(); |
| mgr.unmasquerade(tx); |
| } |
| } |
| |
| @Override |
| protected void process(final ClusterDistributionManager dm) { |
| Throwable thr = null; |
| boolean sendReply = true; |
| DistributedRegion dr = null; |
| TXStateProxy tx = null; |
| |
| try { |
| if (checkCacheClosing(dm) || checkDSClosing(dm)) { |
| InternalCache cache = dm.getCache(); |
| if (cache != null) { |
| thr = cache |
| .getCacheClosedException(String.format("Remote cache is closed: %s", |
| dm.getId())); |
| } else { |
| thr = new CacheClosedException(String.format("Remote cache is closed: %s", |
| dm.getId())); |
| } |
| return; |
| } |
| dr = (DistributedRegion) dm.getCache().getRegion(this.regionPath); |
| if (dr == null) { |
| // if the distributed system is disconnecting, don't send a reply saying |
| // the partitioned region can't be found (bug 36585) |
| thr = new ForceReattemptException(dm.getDistributionManagerId().toString() |
| + ": could not find Distributed region " + regionPath); |
| return; // reply sent in finally block below |
| } |
| thr = UNHANDLED_EXCEPTION; |
| tx = prepForTransaction(dm); |
| sendReply = operateOnDistributedRegion(dm, dr); // need to take care of |
| // it... |
| thr = null; |
| } catch (CancelException se) { |
| // bug 37026: this is too noisy... |
| // throw new CacheClosedException("remote system shutting down"); |
| // thr = se; cache is closed, no point trying to send a reply |
| thr = null; |
| sendReply = false; |
| if (logger.isDebugEnabled()) { |
| logger.debug("shutdown caught, abandoning message: {}", se.getMessage(), se); |
| } |
| } catch (RegionDestroyedException rde) { |
| // [bruce] RDE does not always mean that the sender's region is also |
| // destroyed, so we must send back an exception. If the sender's |
| // region is also destroyed, who cares if we send it an exception |
| if (dr != null && dr.isClosed()) { |
| thr = new ForceReattemptException("Region is destroyed in " + dm.getDistributionManagerId(), |
| rde); |
| } |
| } 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) { |
| logger.debug("{} exception occurred while processing message: {}", this, t.getMessage(), 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(); |
| // log the exception at fine level if there is no reply to the message |
| thr = null; |
| if (sendReply && this.processorId != 0) { |
| if (!checkDSClosing(dm)) { |
| thr = t; |
| } else { |
| // don't pass arbitrary runtime exceptions and errors back if this |
| // cache/vm is closing |
| thr = new ForceReattemptException("Distributed system is disconnecting"); |
| } |
| } |
| if (this.processorId == 0) { |
| logger.debug("{} exception while processing message: {}", this, t.getMessage(), t); |
| } else if (logger.isTraceEnabled(LogMarker.DM_VERBOSE) && (t instanceof RuntimeException)) { |
| logger.trace(LogMarker.DM_VERBOSE, "Exception caught while processing message", t); |
| } |
| } finally { |
| cleanupTransaction(tx); |
| if (sendReply && this.processorId != 0) { |
| ReplyException rex = null; |
| if (thr != null) { |
| // don't transmit the exception if this message was to a listener |
| // and this listener is shutting down |
| boolean excludeException = false; |
| if (!this.functionObject.isHA()) { |
| excludeException = |
| (thr instanceof CacheClosedException || (thr instanceof ForceReattemptException)); |
| } else { |
| excludeException = thr instanceof ForceReattemptException; |
| } |
| if (!excludeException) { |
| rex = new ReplyException(thr); |
| } |
| } |
| // Send the reply if the operateOnPartitionedRegion returned true |
| sendReply(getSender(), this.processorId, dm, rex, null, 0, true, false); |
| } |
| } |
| } |
| |
| protected boolean operateOnDistributedRegion(final ClusterDistributionManager dm, |
| DistributedRegion r) throws ForceReattemptException { |
| if (this.functionObject == null) { |
| ReplyMessage.send(getSender(), this.processorId, |
| new ReplyException(new FunctionException( |
| String.format("Function named %s is not registered to FunctionService", |
| this.functionName))), |
| dm, r.isInternalRegion()); |
| |
| return false; |
| } |
| |
| |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, "FunctionMessage operateOnRegion: {}", r.getFullPath()); |
| } |
| try { |
| r.executeOnRegion(this, this.functionObject, this.args, this.processorId, this.filter, |
| this.isReExecute); |
| if (!this.replyLastMsg && this.functionObject.hasResult()) { |
| ReplyMessage.send(getSender(), this.processorId, |
| new ReplyException(new FunctionException( |
| String.format("The function, %s, did not send last result", |
| functionObject.getId()))), |
| dm, r.isInternalRegion()); |
| return false; |
| } |
| } catch (IOException e) { |
| ReplyMessage |
| .send(getSender(), this.processorId, |
| new ReplyException( |
| "Operation got interrupted due to shutdown in progress on remote VM", e), |
| dm, r.isInternalRegion()); |
| return false; |
| } catch (CancelException sde) { |
| ReplyMessage.send(getSender(), this.processorId, |
| new ReplyException(new ForceReattemptException( |
| "Operation got interrupted due to shutdown in progress on remote VM", sde)), |
| dm, r.isInternalRegion()); |
| return false; |
| } |
| |
| // Unless there was an exception thrown, this message handles sending the |
| // response |
| return false; |
| } |
| |
| /** |
| * check to see if the cache is closing |
| */ |
| private boolean checkCacheClosing(ClusterDistributionManager dm) { |
| InternalCache cache = dm.getCache(); |
| return cache == null || cache.getCancelCriterion().isCancelInProgress(); |
| } |
| |
| /** |
| * check to see if the distributed system is closing |
| * |
| * @return true if the distributed system is closing |
| */ |
| private boolean checkDSClosing(ClusterDistributionManager dm) { |
| InternalDistributedSystem ds = dm.getSystem(); |
| return (ds == null || ds.isDisconnecting()); |
| } |
| |
| @Override |
| public int getProcessorId() { |
| return this.processorId; |
| } |
| |
| @Override |
| public int getDSFID() { |
| return DR_FUNCTION_STREAMING_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| |
| short flags = in.readShort(); |
| if ((flags & HAS_PROCESSOR_ID) != 0) { |
| this.processorId = in.readInt(); |
| ReplyProcessor21.setMessageRPId(this.processorId); |
| } |
| if ((flags & HAS_TX_ID) != 0) |
| this.txUniqId = in.readInt(); |
| if ((flags & HAS_TX_MEMBERID) != 0) { |
| this.txMemberId = DataSerializer.readObject(in); |
| } |
| |
| Object object = DataSerializer.readObject(in); |
| if (object instanceof String) { |
| this.isFnSerializationReqd = false; |
| this.functionObject = FunctionService.getFunction((String) object); |
| if (this.functionObject == null) { |
| this.functionName = (String) object; |
| } |
| } else { |
| this.functionObject = (Function) object; |
| this.isFnSerializationReqd = true; |
| } |
| this.args = (Serializable) DataSerializer.readObject(in); |
| this.filter = (HashSet) DataSerializer.readHashSet(in); |
| this.regionPath = DataSerializer.readString(in); |
| this.isReExecute = (flags & IS_REEXECUTE) != 0; |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| |
| short flags = 0; |
| if (this.processorId != 0) |
| flags |= HAS_PROCESSOR_ID; |
| if (this.txUniqId != TXManagerImpl.NOTX) |
| flags |= HAS_TX_ID; |
| if (this.txMemberId != null) |
| flags |= HAS_TX_MEMBERID; |
| if (this.isReExecute) |
| flags |= IS_REEXECUTE; |
| out.writeShort(flags); |
| if (this.processorId != 0) |
| out.writeInt(this.processorId); |
| if (this.txUniqId != TXManagerImpl.NOTX) |
| out.writeInt(this.txUniqId); |
| if (this.txMemberId != null) { |
| DataSerializer.writeObject(this.txMemberId, out); |
| } |
| |
| if (this.isFnSerializationReqd) { |
| DataSerializer.writeObject(this.functionObject, out); |
| } else { |
| DataSerializer.writeObject(functionObject.getId(), out); |
| } |
| DataSerializer.writeObject(this.args, out); |
| DataSerializer.writeHashSet((HashSet) this.filter, out); |
| DataSerializer.writeString(this.regionPath, out); |
| } |
| |
| public synchronized boolean sendReplyForOneResult(DistributionManager dm, Object oneResult, |
| boolean lastResult, boolean sendResultsInOrder) |
| throws CacheException, ForceReattemptException, InterruptedException { |
| if (this.replyLastMsg) { |
| return false; |
| } |
| if (Thread.interrupted()) |
| throw new InterruptedException(); |
| int msgNum = this.replyMsgNum; |
| this.replyLastMsg = lastResult; |
| |
| sendReply(getSender(), this.processorId, dm, null, oneResult, msgNum, lastResult, |
| sendResultsInOrder); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Sending reply message count: {} to co-ordinating node", replyMsgNum); |
| } |
| this.replyMsgNum++; |
| return false; |
| } |
| |
| protected void sendReply(InternalDistributedMember member, int procId, DistributionManager dm, |
| ReplyException ex, Object result, int msgNum, boolean lastResult, |
| boolean sendResultsInOrder) { |
| // if there was an exception, then throw out any data |
| if (ex != null) { |
| this.result = null; |
| this.replyMsgNum = 0; |
| this.replyLastMsg = true; |
| } |
| if (sendResultsInOrder) { |
| FunctionStreamingOrderedReplyMessage.send(member, procId, ex, dm, result, msgNum, lastResult); |
| } else { |
| FunctionStreamingReplyMessage.send(member, procId, ex, dm, result, msgNum, lastResult); |
| } |
| } |
| |
| @Override |
| public int getProcessorType() { |
| return OperationExecutors.REGION_FUNCTION_EXECUTION_EXECUTOR; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TransactionMessage#canStartRemoteTransaction() |
| */ |
| @Override |
| public boolean canStartRemoteTransaction() { |
| return true; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.geode.internal.cache.TransactionMessage#getTXUniqId() |
| */ |
| @Override |
| public int getTXUniqId() { |
| return txUniqId; |
| } |
| |
| @Override |
| public InternalDistributedMember getMemberToMasqueradeAs() { |
| if (txMemberId == null) { |
| return getSender(); |
| } else { |
| return txMemberId; |
| } |
| } |
| |
| @Override |
| public InternalDistributedMember getTXOriginatorClient() { |
| // TODO Auto-generated method stub |
| return null; |
| } |
| |
| @Override |
| public boolean canParticipateInTransaction() { |
| return true; |
| } |
| |
| @Override |
| public boolean isTransactionDistributed() { |
| return false; |
| } |
| } |
