| /* |
| * 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 static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT; |
| import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.Vector; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.InvalidDeltaException; |
| import org.apache.geode.InvalidVersionException; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.annotations.internal.MutableForTesting; |
| import org.apache.geode.cache.CacheEvent; |
| import org.apache.geode.cache.DiskAccessException; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.EntryOperation; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.query.internal.cq.CqService; |
| import org.apache.geode.cache.query.internal.cq.ServerCQ; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DirectReplyProcessor; |
| 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.MessageWithReply; |
| 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.ReplySender; |
| import org.apache.geode.distributed.internal.SerialDistributionMessage; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.CopyOnWriteHashSet; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile; |
| import org.apache.geode.internal.cache.EntryEventImpl.OldValueImporter; |
| import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo; |
| import org.apache.geode.internal.cache.LocalRegion.InitializationLevel; |
| import org.apache.geode.internal.cache.UpdateOperation.UpdateMessage; |
| import org.apache.geode.internal.cache.partitioned.Bucket; |
| import org.apache.geode.internal.cache.partitioned.PartitionMessage; |
| import org.apache.geode.internal.cache.persistence.PersistentMemberID; |
| import org.apache.geode.internal.cache.tier.MessageType; |
| import org.apache.geode.internal.cache.tx.RemoteOperationMessage; |
| import org.apache.geode.internal.cache.versions.DiskVersionTag; |
| import org.apache.geode.internal.cache.versions.RegionVersionVector; |
| import org.apache.geode.internal.cache.versions.VersionSource; |
| import org.apache.geode.internal.cache.versions.VersionTag; |
| import org.apache.geode.internal.logging.LogService; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.offheap.Releasable; |
| import org.apache.geode.internal.offheap.StoredObject; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.offheap.annotations.Unretained; |
| import org.apache.geode.internal.sequencelog.EntryLogger; |
| import org.apache.geode.internal.util.DelayedAction; |
| |
| public abstract class DistributedCacheOperation { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| @MutableForTesting |
| public static double LOSS_SIMULATION_RATIO = 0; // test hook |
| |
| @MutableForTesting |
| public static Random LOSS_SIMULATION_GENERATOR; |
| |
| @MutableForTesting |
| public static long SLOW_DISTRIBUTION_MS = 0; // test hook |
| |
| // constants used in subclasses and distribution messages |
| // should use enum in source level 1.5+ |
| |
| /** |
| * Deserialization policy: do not deserialize (for byte array, null or cases where the value |
| * should stay serialized) |
| * |
| * @since GemFire 5.7 |
| */ |
| public static final byte DESERIALIZATION_POLICY_NONE = (byte) 0; |
| |
| /** |
| * Deserialization policy: deserialize lazily (for all other objects) |
| * |
| * @since GemFire 5.7 |
| */ |
| public static final byte DESERIALIZATION_POLICY_LAZY = (byte) 2; |
| |
| /** |
| * @param deserializationPolicy must be one of the following: DESERIALIZATION_POLICY_NONE, |
| * DESERIALIZATION_POLICY_LAZY. |
| */ |
| public static void writeValue(final byte deserializationPolicy, final Object vObj, |
| final byte[] vBytes, final DataOutput out) throws IOException { |
| if (vObj != null) { |
| if (deserializationPolicy == DESERIALIZATION_POLICY_NONE) { |
| // We only have NONE with a vObj when vObj is off-heap and not serialized. |
| StoredObject so = (StoredObject) vObj; |
| assert !so.isSerialized(); |
| so.sendAsByteArray(out); |
| } else { // LAZY |
| DataSerializer.writeObjectAsByteArray(vObj, out); |
| } |
| } else { |
| DataSerializer.writeByteArray(vBytes, out); |
| } |
| } |
| |
| // static values for oldValueIsObject |
| public static final byte VALUE_IS_BYTES = 0; |
| public static final byte VALUE_IS_SERIALIZED_OBJECT = 1; |
| public static final byte VALUE_IS_OBJECT = 2; |
| |
| /** |
| * Given a VALUE_IS_* constant convert and return the corresponding DESERIALIZATION_POLICY_*. |
| */ |
| public static byte valueIsToDeserializationPolicy(boolean oldValueIsSerialized) { |
| if (!oldValueIsSerialized) |
| return DESERIALIZATION_POLICY_NONE; |
| return DESERIALIZATION_POLICY_LAZY; |
| } |
| |
| |
| public static final byte DESERIALIZATION_POLICY_NUMBITS = |
| DistributionMessage.getNumBits(DESERIALIZATION_POLICY_LAZY); |
| |
| public static final short DESERIALIZATION_POLICY_END = |
| (short) (1 << DESERIALIZATION_POLICY_NUMBITS); |
| |
| public static final short DESERIALIZATION_POLICY_MASK = (short) (DESERIALIZATION_POLICY_END - 1); |
| |
| @MutableForTesting |
| public static boolean testSendingOldValues; |
| |
| protected InternalCacheEvent event; |
| |
| protected CacheOperationReplyProcessor processor = null; |
| |
| protected Set departedMembers; |
| |
| protected Set originalRecipients; |
| |
| @MutableForTesting |
| static Runnable internalBeforePutOutgoing; |
| |
| public static String deserializationPolicyToString(byte policy) { |
| switch (policy) { |
| case DESERIALIZATION_POLICY_NONE: |
| return "NONE"; |
| case DESERIALIZATION_POLICY_LAZY: |
| return "LAZY"; |
| default: |
| throw new AssertionError("unknown deserialization policy"); |
| } |
| } |
| |
| /** Creates a new instance of DistributedCacheOperation */ |
| public DistributedCacheOperation(CacheEvent event) { |
| this.event = (InternalCacheEvent) event; |
| } |
| |
| /** |
| * Return true if this operation needs to check for reliable delivery. Return false if not. |
| * Currently the only case it doesn't need to be is a DestroyRegionOperation doing a "local" |
| * destroy. |
| * |
| * @since GemFire 5.0 |
| */ |
| boolean isOperationReliable() { |
| Operation op = this.event.getOperation(); |
| if (!op.isRegionDestroy()) { |
| return true; |
| } |
| if (op.isDistributed()) { |
| return true; |
| } |
| // must be a region destroy that is "local" which means |
| // Region.localDestroyRegion or Region.close or Cache.clsoe |
| // none of these should do reliability checks |
| return false; |
| } |
| |
| public boolean supportsDirectAck() { |
| // force use of shared connection if we're already in a secondary |
| // thread-owned reader thread. See bug #49565. Also see Connection#processNIOBuffer |
| // int dominoCount = org.apache.geode.internal.tcp.Connection.getDominoCount(); |
| // return dominoCount < 2; |
| return true; |
| } |
| |
| /** |
| * returns true if multicast can be used for this operation. The default is true. |
| */ |
| public boolean supportsMulticast() { |
| return true; |
| } |
| |
| /** returns true if the receiver can distribute during cache closure */ |
| public boolean canBeSentDuringShutdown() { |
| return getRegion().isUsedForPartitionedRegionAdmin(); |
| } |
| |
| /** |
| * returns true if adjunct messaging (piggybacking) is allowed for this operation. Region-oriented |
| * operations typically do not allow adjunct messaging, while Entry-oriented operations do. The |
| * default implementation returns true. |
| */ |
| protected boolean supportsAdjunctMessaging() { |
| return true; |
| } |
| |
| /** |
| * returns true if this operation supports propagation of delta values instead of full changes |
| */ |
| protected boolean supportsDeltaPropagation() { |
| return false; |
| } |
| |
| /** does this operation change region content? */ |
| public boolean containsRegionContentChange() { |
| return true; |
| } |
| |
| @MutableForTesting |
| public static volatile DelayedAction test_InvalidVersionAction; |
| |
| /** |
| * region's distribution advisor marked that a distribution is about to start, then distribute. It |
| * returns a token, which is view version. Return -1 means the method did not succeed. This method |
| * must be invoked before toDistribute(). This method should pair with endOperation() in |
| * try/finally block. |
| */ |
| public long startOperation() { |
| DistributedRegion region = getRegion(); |
| long viewVersion = -1; |
| try { |
| if (this.containsRegionContentChange()) { |
| viewVersion = region.getDistributionAdvisor().startOperation(); |
| } |
| if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) { |
| logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "dispatching operation in view version {}", |
| viewVersion); |
| } |
| try { |
| _distribute(); |
| } catch (InvalidVersionException e) { |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, |
| "PutAll failed since versions were missing; retrying", e); |
| } |
| |
| if (test_InvalidVersionAction != null) { |
| test_InvalidVersionAction.run(); |
| } |
| _distribute(); |
| } |
| } catch (RuntimeException | Error e) { |
| endOperation(viewVersion); |
| throw e; |
| } |
| return viewVersion; |
| } |
| |
| /** |
| * region's distribution advisor marked that a distribution is ended. This method should pair with |
| * startOperation in try/finally block. |
| */ |
| public void endOperation(long viewVersion) { |
| DistributedRegion region = getRegion(); |
| if (viewVersion != -1) { |
| region.getDistributionAdvisor().endOperation(viewVersion); |
| if (logger.isTraceEnabled()) { |
| logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, |
| "done dispatching operation in view version {}", viewVersion); |
| } |
| } |
| } |
| |
| /** |
| * Distribute a cache operation to other members of the distributed system. This method determines |
| * who the recipients are and handles careful delivery of the operation to those members. |
| */ |
| public void distribute() { |
| long token = -1; |
| try { |
| token = startOperation(); |
| } finally { |
| endOperation(token); |
| } |
| } |
| |
| /** |
| * About to distribute a cache operation to other members of the distributed system. This method |
| * determines who the recipients are and handles careful delivery of the operation to those |
| * members. This method should wrapped by startOperation() and endOperation() in try/finally |
| * block. |
| */ |
| protected void _distribute() { |
| DistributedRegion region = getRegion(); |
| DistributionManager mgr = region.getDistributionManager(); |
| boolean reliableOp = isOperationReliable() && region.requiresReliabilityCheck(); |
| |
| if (SLOW_DISTRIBUTION_MS > 0) { // test hook |
| try { |
| Thread.sleep(SLOW_DISTRIBUTION_MS); |
| } catch (InterruptedException ignore) { |
| Thread.currentThread().interrupt(); |
| } |
| SLOW_DISTRIBUTION_MS = 0; |
| } |
| |
| boolean isPutAll = (this instanceof DistributedPutAllOperation); |
| boolean isRemoveAll = (this instanceof DistributedRemoveAllOperation); |
| |
| try { |
| // Recipients with CacheOp |
| Set<InternalDistributedMember> recipients = getRecipients(); |
| Map<InternalDistributedMember, PersistentMemberID> persistentIds = null; |
| if (region.getDataPolicy().withPersistence()) { |
| persistentIds = region.getDistributionAdvisor().adviseInitializedPersistentMembers(); |
| } |
| |
| // some members requiring old value are also in the cache op recipients set |
| Set needsOldValueInCacheOp = Collections.emptySet(); |
| |
| // set client routing information into the event |
| boolean routingComputed = false; |
| FilterRoutingInfo filterRouting = null; |
| // recipients that will get a cacheop msg and also a PR message |
| Set twoMessages = Collections.emptySet(); |
| if (region.isUsedForPartitionedRegionBucket()) { |
| twoMessages = ((Bucket) region).getBucketAdvisor().adviseRequiresTwoMessages(); |
| routingComputed = true; |
| filterRouting = getRecipientFilterRouting(recipients); |
| if (filterRouting != null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Computed this filter routing: {}", filterRouting); |
| } |
| } |
| } |
| |
| // some members need PR notification of the change for client/wan |
| // notification |
| Set adjunctRecipients = Collections.emptySet(); |
| |
| // Partitioned region listener notification messages piggyback on this |
| // operation's replyprocessor and need to be sent at the same time as |
| // the operation's message |
| if (this.supportsAdjunctMessaging() && region.isUsedForPartitionedRegionBucket()) { |
| BucketRegion br = (BucketRegion) region; |
| adjunctRecipients = getAdjunctReceivers(br, recipients, twoMessages, filterRouting); |
| } |
| |
| EntryEventImpl entryEvent = event.getOperation().isEntry() ? getEvent() : null; |
| |
| if (entryEvent != null && entryEvent.hasOldValue()) { |
| if (testSendingOldValues) { |
| needsOldValueInCacheOp = new HashSet(recipients); |
| } else { |
| needsOldValueInCacheOp = |
| region.getCacheDistributionAdvisor().adviseRequiresOldValueInCacheOp(); |
| } |
| recipients.removeAll(needsOldValueInCacheOp); |
| } |
| |
| Set cachelessNodes = Collections.emptySet(); |
| Set adviseCacheServers; |
| Set<InternalDistributedMember> cachelessNodesWithNoCacheServer = Collections.emptySet(); |
| if (region.getDistributionConfig().getDeltaPropagation() && this.supportsDeltaPropagation()) { |
| cachelessNodes = region.getCacheDistributionAdvisor().adviseEmptys(); |
| if (!cachelessNodes.isEmpty()) { |
| List list = new ArrayList(cachelessNodes); |
| for (Object member : cachelessNodes) { |
| if (!recipients.contains(member) || adjunctRecipients.contains(member)) { |
| // Don't include those originally excluded. |
| list.remove(member); |
| } |
| } |
| cachelessNodes.clear(); |
| recipients.removeAll(list); |
| cachelessNodes.addAll(list); |
| } |
| if (!cachelessNodes.isEmpty()) { |
| cachelessNodesWithNoCacheServer = new HashSet<>(cachelessNodes); |
| adviseCacheServers = region.getCacheDistributionAdvisor().adviseCacheServers(); |
| cachelessNodesWithNoCacheServer.removeAll(adviseCacheServers); |
| } |
| } |
| |
| if (recipients.isEmpty() && adjunctRecipients.isEmpty() && needsOldValueInCacheOp.isEmpty() |
| && cachelessNodes.isEmpty()) { |
| if (region.isInternalRegion()) { |
| if (logger.isDebugEnabled() || logger.isTraceEnabled()) { |
| if (mgr.getNormalDistributionManagerIds().size() > 1) { |
| // suppress this msg if we are the only member. |
| if (logger.isTraceEnabled()) { |
| logger.trace("<No Recipients> {}", this); |
| } |
| } else { |
| // suppress this msg if we are the only member. |
| if (logger.isDebugEnabled()) { |
| logger.debug("<No Recipients> {}", this); |
| } |
| } |
| } |
| } |
| if (!reliableOp || region.isNoDistributionOk()) { |
| // nothing needs be done in this case |
| } else { |
| region.handleReliableDistribution(Collections.emptySet()); |
| } |
| |
| // compute local client routing before waiting for an ack only for a bucket |
| if (region.isUsedForPartitionedRegionBucket()) { |
| FilterInfo filterInfo = getLocalFilterRouting(filterRouting); |
| this.event.setLocalFilterInfo(filterInfo); |
| } |
| |
| } else { |
| boolean directAck = false; |
| boolean useMulticast = region.getMulticastEnabled() |
| && region.getSystem().getConfig().getMcastPort() != 0 && this.supportsMulticast(); |
| boolean shouldAck = shouldAck(); |
| |
| if (shouldAck) { |
| if (this.supportsDirectAck() && adjunctRecipients.isEmpty()) { |
| if (region.getSystem().threadOwnsResources()) { |
| directAck = true; |
| } |
| } |
| } |
| // don't send to the sender of a remote-operation-message. Those messages send |
| // their own response. fixes bug #45973 |
| if (entryEvent != null) { |
| RemoteOperationMessage rmsg = entryEvent.getRemoteOperationMessage(); |
| if (rmsg != null) { |
| recipients.remove(rmsg.getSender()); |
| useMulticast = false; // bug #45106: can't mcast or the sender of the one-hop op will |
| // get it |
| } |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("recipients for {}: {} with adjunct messages to: {}", this, recipients, |
| adjunctRecipients); |
| } |
| |
| if (shouldAck) { |
| // adjunct messages are sent using the same reply processor, so |
| // add them to the processor's membership set |
| Collection waitForMembers = null; |
| if (recipients.size() > 0 && adjunctRecipients.size() == 0 && cachelessNodes.isEmpty()) { // the |
| // common |
| // case |
| waitForMembers = recipients; |
| } else if (!cachelessNodes.isEmpty()) { |
| waitForMembers = new HashSet(recipients); |
| waitForMembers.addAll(cachelessNodes); |
| } else { |
| // note that we use a Vector instead of a Set for the responders |
| // collection |
| // because partitioned regions sometimes send both a regular cache |
| // operation and a partitioned-region notification message to the |
| // same recipient |
| waitForMembers = new Vector(recipients); |
| waitForMembers.addAll(adjunctRecipients); |
| waitForMembers.addAll(needsOldValueInCacheOp); |
| waitForMembers.addAll(cachelessNodes); |
| } |
| if (DistributedCacheOperation.LOSS_SIMULATION_RATIO != 0.0) { |
| if (LOSS_SIMULATION_GENERATOR == null) { |
| LOSS_SIMULATION_GENERATOR = new Random(this.hashCode()); |
| } |
| if ((LOSS_SIMULATION_GENERATOR.nextInt(100) * 1.0 / 100.0) < LOSS_SIMULATION_RATIO) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("loss simulation is inhibiting message transmission to {}", |
| recipients); |
| } |
| waitForMembers.removeAll(recipients); |
| recipients = Collections.emptySet(); |
| } |
| } |
| if (reliableOp) { |
| this.departedMembers = new HashSet(); |
| this.processor = new ReliableCacheReplyProcessor(region.getSystem(), waitForMembers, |
| this.departedMembers); |
| } else { |
| this.processor = new CacheOperationReplyProcessor(region.getSystem(), waitForMembers); |
| } |
| } |
| |
| Set failures = null; |
| CacheOperationMessage msg = createMessage(); |
| initMessage(msg, this.processor); |
| |
| if (DistributedCacheOperation.internalBeforePutOutgoing != null) { |
| DistributedCacheOperation.internalBeforePutOutgoing.run(); |
| } |
| |
| if (processor != null && msg.isSevereAlertCompatible()) { |
| this.processor.enableSevereAlertProcessing(); |
| // if this message is distributing for a partitioned region message, |
| // we can't wait as long as the full ack-severe-alert-threshold or |
| // the sender might kick us out of the system before we can get an ack |
| // back |
| DistributedRegion r = getRegion(); |
| if (r.isUsedForPartitionedRegionBucket() && event.getOperation().isEntry()) { |
| PartitionMessage pm = ((EntryEventImpl) event).getPartitionMessage(); |
| if (pm != null && pm.getSender() != null |
| && !pm.getSender().equals(r.getDistributionManager().getDistributionManagerId())) { |
| // PR message sent by another member |
| ReplyProcessor21.setShortSevereAlertProcessing(true); |
| } |
| } |
| } |
| |
| msg.setMulticast(useMulticast); |
| msg.directAck = directAck; |
| if (region.isUsedForPartitionedRegionBucket()) { |
| if (!isPutAll && !isRemoveAll && filterRouting != null |
| && filterRouting.hasMemberWithFilterInfo()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Setting filter information for message to {}", filterRouting); |
| } |
| msg.filterRouting = filterRouting; |
| } |
| } else if (!routingComputed) { |
| msg.needsRouting = true; |
| } |
| |
| initProcessor(processor, msg); |
| |
| if (region.cache.isClosed() && !canBeSentDuringShutdown()) { |
| throw region.cache.getCacheClosedException( |
| "The cache has been closed", |
| null); |
| } |
| |
| msg.setRecipients(recipients); |
| failures = mgr.putOutgoing(msg); |
| |
| // distribute to members needing the old value now |
| if (needsOldValueInCacheOp.size() > 0) { |
| msg.appendOldValueToMessage((EntryEventImpl) this.event); |
| msg.resetRecipients(); |
| msg.setRecipients(needsOldValueInCacheOp); |
| Set newFailures = mgr.putOutgoing(msg); |
| if (newFailures != null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Failed sending ({}) to {}", msg, newFailures); |
| } |
| if (failures != null && failures.size() > 0) { |
| failures.addAll(newFailures); |
| } else { |
| failures = newFailures; |
| } |
| } |
| } |
| |
| if (cachelessNodes.size() > 0) { |
| cachelessNodes.removeAll(cachelessNodesWithNoCacheServer); |
| if (cachelessNodes.size() > 0) { |
| msg.resetRecipients(); |
| msg.setRecipients(cachelessNodes); |
| msg.setSendDelta(false); |
| Set newFailures = mgr.putOutgoing(msg); |
| if (newFailures != null) { |
| if (failures != null && failures.size() > 0) { |
| failures.addAll(newFailures); |
| } else { |
| failures = newFailures; |
| } |
| } |
| } |
| |
| if (!cachelessNodesWithNoCacheServer.isEmpty()) { |
| msg.resetRecipients(); |
| msg.setRecipients(cachelessNodesWithNoCacheServer); |
| msg.setSendDelta(false); |
| ((UpdateMessage) msg).setSendDeltaWithFullValue(false); |
| Set newFailures = mgr.putOutgoing(msg); |
| if (newFailures != null) { |
| if (failures != null && failures.size() > 0) { |
| failures.addAll(newFailures); |
| } else { |
| failures = newFailures; |
| } |
| } |
| // Add it back for size calculation ahead |
| cachelessNodes.addAll(cachelessNodesWithNoCacheServer); |
| } |
| } |
| |
| if (failures != null && !failures.isEmpty() && logger.isDebugEnabled()) { |
| logger.debug("Failed sending ({}) to {} while processing event:{}", msg, failures, event); |
| } |
| |
| Set<InternalDistributedMember> adjunctRecipientsWithNoCacheServer = Collections.emptySet(); |
| // send partitioned region listener notification messages now |
| if (!adjunctRecipients.isEmpty()) { |
| if (cachelessNodes.size() > 0) { |
| // add non-delta recipients back into the set for adjunct |
| // calculations |
| if (recipients.isEmpty()) { |
| recipients = cachelessNodes; |
| } else { |
| recipients.addAll(cachelessNodes); |
| } |
| } |
| adjunctRecipientsWithNoCacheServer = new HashSet<>(adjunctRecipients); |
| adviseCacheServers = ((Bucket) region).getPartitionedRegion() |
| .getCacheDistributionAdvisor().adviseCacheServers(); |
| adjunctRecipientsWithNoCacheServer.removeAll(adviseCacheServers); |
| |
| if (isPutAll) { |
| ((BucketRegion) region).performPutAllAdjunctMessaging((DistributedPutAllOperation) this, |
| recipients, adjunctRecipients, filterRouting, this.processor); |
| } else if (isRemoveAll) { |
| ((BucketRegion) region).performRemoveAllAdjunctMessaging( |
| (DistributedRemoveAllOperation) this, recipients, adjunctRecipients, filterRouting, |
| this.processor); |
| } else { |
| boolean calculateDelta = |
| adjunctRecipientsWithNoCacheServer.size() < adjunctRecipients.size(); |
| adjunctRecipients.removeAll(adjunctRecipientsWithNoCacheServer); |
| if (!adjunctRecipients.isEmpty()) { |
| ((BucketRegion) region).performAdjunctMessaging(getEvent(), recipients, |
| adjunctRecipients, filterRouting, this.processor, calculateDelta, true); |
| } |
| if (!adjunctRecipientsWithNoCacheServer.isEmpty()) { |
| ((BucketRegion) region).performAdjunctMessaging(getEvent(), recipients, |
| adjunctRecipientsWithNoCacheServer, filterRouting, this.processor, calculateDelta, |
| false); |
| } |
| } |
| } |
| |
| // compute local client routing before waiting for an ack only for a bucket |
| if (region.isUsedForPartitionedRegionBucket()) { |
| FilterInfo filterInfo = getLocalFilterRouting(filterRouting); |
| event.setLocalFilterInfo(filterInfo); |
| } |
| |
| waitForAckIfNeeded(msg, persistentIds); |
| |
| if (/* msg != null && */reliableOp) { |
| Set successfulRecips = new HashSet(recipients); |
| successfulRecips.addAll(cachelessNodes); |
| successfulRecips.addAll(needsOldValueInCacheOp); |
| if (failures != null && !failures.isEmpty()) { |
| successfulRecips.removeAll(failures); |
| } |
| if (departedMembers != null) { |
| successfulRecips.removeAll(departedMembers); |
| } |
| region.handleReliableDistribution(successfulRecips); |
| } |
| } |
| |
| if (region.isUsedForPartitionedRegionBucket() && filterRouting != null) { |
| removeDestroyTokensFromCqResultKeys(filterRouting); |
| } |
| |
| } catch (CancelException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("distribution of message aborted by shutdown: {}", this); |
| } |
| throw e; |
| } catch (RuntimeException e) { |
| logger.info(String.format("Exception occurred while processing %s", this), |
| e); |
| throw e; |
| } finally { |
| ReplyProcessor21.setShortSevereAlertProcessing(false); |
| } |
| } |
| |
| /** |
| * Cleanup destroyed events in CQ result cache for remote CQs. While maintaining the CQ results |
| * key caching. the destroy event keys are marked as destroyed instead of removing them, this is |
| * to take care, arrival of duplicate events. The key marked as destroyed are removed after the |
| * event is placed in clients HAQueue or distributed to the peers. |
| * |
| * This is similar to CacheClientNotifier.removeDestroyTokensFromCqResultKeys() where the |
| * destroyed events for local CQs are handled. |
| */ |
| private void removeDestroyTokensFromCqResultKeys(FilterRoutingInfo filterRouting) { |
| for (InternalDistributedMember m : filterRouting.getMembers()) { |
| FilterInfo filterInfo = filterRouting.getFilterInfo(m); |
| if (filterInfo.getCQs() == null) { |
| continue; |
| } |
| |
| CacheProfile cf = (CacheProfile) ((Bucket) getRegion()).getPartitionedRegion() |
| .getCacheDistributionAdvisor().getProfile(m); |
| |
| if (cf == null || cf.filterProfile == null || cf.filterProfile.isLocalProfile() |
| || cf.filterProfile.getCqMap().isEmpty()) { |
| continue; |
| } |
| |
| for (Object value : cf.filterProfile.getCqMap().values()) { |
| ServerCQ cq = (ServerCQ) value; |
| |
| for (Map.Entry<Long, Integer> e : filterInfo.getCQs().entrySet()) { |
| Long cqID = e.getKey(); |
| // For the CQs satisfying the event with destroy CQEvent, remove |
| // the entry form CQ cache. |
| if (cq != null && cq.getFilterID() != null && cq.getFilterID().equals(cqID) |
| && (e.getValue().equals(MessageType.LOCAL_DESTROY))) { |
| cq.removeFromCqResultKeys(((EntryOperation) event).getKey(), true); |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * Get the adjunct receivers for a partitioned region operation |
| * |
| * @param br the PR bucket |
| * @param cacheOpReceivers the receivers of the CacheOperationMessage for this op |
| * @param twoMessages PR members that are creating the bucket and need both cache op and adjunct |
| * messages |
| * @param routing client routing information |
| */ |
| Set getAdjunctReceivers(BucketRegion br, Set cacheOpReceivers, Set twoMessages, |
| FilterRoutingInfo routing) { |
| return br.getAdjunctReceivers(this.getEvent(), cacheOpReceivers, twoMessages, routing); |
| } |
| |
| /** |
| * perform any operation-specific initialization on the given reply processor |
| */ |
| protected void initProcessor(CacheOperationReplyProcessor p, CacheOperationMessage msg) { |
| // nothing to do here - see UpdateMessage |
| } |
| |
| protected void waitForAckIfNeeded(CacheOperationMessage msg, |
| Map<InternalDistributedMember, PersistentMemberID> persistentIds) { |
| if (this.processor == null) { |
| return; |
| } |
| try { |
| // keep waiting even if interrupted |
| try { |
| this.processor.waitForRepliesUninterruptibly(); |
| Set<InternalDistributedMember> closedMembers = this.processor.closedMembers.getSnapshot(); |
| handleClosedMembers(closedMembers, persistentIds); |
| } catch (ReplyException e) { |
| if (this instanceof DestroyRegionOperation) { |
| logger.fatal("waitForAckIfNeeded: exception", e); |
| } |
| e.handleCause(); |
| } |
| } finally { |
| this.processor = null; |
| } |
| } |
| |
| private void handleClosedMembers(Set<InternalDistributedMember> closedMembers, |
| Map<InternalDistributedMember, PersistentMemberID> persistentIds) { |
| if (persistentIds == null) { |
| return; |
| } |
| |
| for (InternalDistributedMember member : closedMembers) { |
| PersistentMemberID persistentId = persistentIds.get(member); |
| if (persistentId != null) { |
| // Fix for bug 42142 - In order for recovery to work, |
| // we must either |
| // 1) persistent the region operation successfully on the peer |
| // 2) record that the peer is offline |
| // or |
| // 3) fail the operation |
| |
| // if we have started to shutdown, we don't want to mark the peer |
| // as offline, or we will think we have newer data when in fact we don't |
| getRegion().getCancelCriterion().checkCancelInProgress(null); |
| |
| // Otherwise, mark the peer as offline, because it didn't complete |
| // the operation. |
| getRegion().getPersistenceAdvisor().markMemberOffline(member, persistentId); |
| } |
| } |
| } |
| |
| protected boolean shouldAck() { |
| return getRegion().scope.isAck(); |
| } |
| |
| protected DistributedRegion getRegion() { |
| return (DistributedRegion) this.event.getRegion(); |
| } |
| |
| protected EntryEventImpl getEvent() { |
| return (EntryEventImpl) this.event; |
| } |
| |
| protected Set getRecipients() { |
| CacheDistributionAdvisor advisor = getRegion().getCacheDistributionAdvisor(); |
| this.originalRecipients = advisor.adviseCacheOp(); |
| return this.originalRecipients; |
| } |
| |
| protected FilterRoutingInfo getRecipientFilterRouting(Set cacheOpRecipients) { |
| LocalRegion region = getRegion(); |
| if (!region.isUsedForPartitionedRegionBucket()) { |
| return null; |
| } |
| CacheDistributionAdvisor advisor; |
| advisor = region.getPartitionedRegion().getCacheDistributionAdvisor(); |
| return advisor.adviseFilterRouting(this.event, cacheOpRecipients); |
| } |
| |
| /** |
| * @param frInfo the filter routing computed for distribution to peers |
| * @return the filter routing computed for distribution to clients of this process |
| */ |
| protected FilterInfo getLocalFilterRouting(FilterRoutingInfo frInfo) { |
| FilterProfile fp = getRegion().getFilterProfile(); |
| if (fp == null) { |
| return null; |
| } |
| FilterRoutingInfo fri = fp.getFilterRoutingInfoPart2(frInfo, this.event); |
| if (fri == null) { |
| return null; |
| } |
| return fri.getLocalFilterInfo(); |
| } |
| |
| protected abstract CacheOperationMessage createMessage(); |
| |
| protected void initMessage(CacheOperationMessage msg, DirectReplyProcessor p) { |
| msg.regionPath = getRegion().getFullPath(); |
| msg.processorId = p == null ? 0 : p.getProcessorId(); |
| msg.processor = p; |
| if (this.event.getOperation().isEntry()) { |
| EntryEventImpl entryEvent = getEvent(); |
| msg.callbackArg = entryEvent.getRawCallbackArgument(); |
| msg.possibleDuplicate = entryEvent.isPossibleDuplicate(); |
| |
| VersionTag tag = entryEvent.getVersionTag(); |
| msg.setInhibitNotificationsBit(entryEvent.inhibitAllNotifications()); |
| if (tag != null && tag.hasValidVersion()) { |
| msg.setVersionTag(tag); |
| } |
| |
| } else { |
| msg.callbackArg = ((RegionEventImpl) this.event).getRawCallbackArgument(); |
| } |
| msg.op = this.event.getOperation(); |
| msg.owner = this; |
| msg.regionAllowsConflation = getRegion().getEnableAsyncConflation(); |
| |
| } |
| |
| @Override |
| public String toString() { |
| String cname = getClass().getName().substring(getClass().getPackage().getName().length() + 1); |
| return cname + "(" + this.event + ")"; |
| } |
| |
| /** |
| * Add an internal callback which is run before the CacheOperationMessage is distributed with |
| * dm.putOutgoing. |
| */ |
| public static void setBeforePutOutgoing(Runnable beforePutOutgoing) { |
| internalBeforePutOutgoing = beforePutOutgoing; |
| } |
| |
| public abstract static class CacheOperationMessage extends SerialDistributionMessage |
| implements MessageWithReply, DirectReplyMessage, OldValueImporter { |
| |
| protected static final short POSSIBLE_DUPLICATE_MASK = POS_DUP; |
| protected static final short OLD_VALUE_MASK = DistributionMessage.UNRESERVED_FLAGS_START; |
| protected static final short DIRECT_ACK_MASK = (OLD_VALUE_MASK << 1); |
| protected static final short FILTER_INFO_MASK = (DIRECT_ACK_MASK << 1); |
| protected static final short CALLBACK_ARG_MASK = (FILTER_INFO_MASK << 1); |
| protected static final short DELTA_MASK = (CALLBACK_ARG_MASK << 1); |
| protected static final short NEEDS_ROUTING_MASK = (DELTA_MASK << 1); |
| protected static final short VERSION_TAG_MASK = (NEEDS_ROUTING_MASK << 1); |
| protected static final short PERSISTENT_TAG_MASK = (VERSION_TAG_MASK << 1); |
| protected static final short UNRESERVED_FLAGS_START = (PERSISTENT_TAG_MASK << 1); |
| |
| private static final int INHIBIT_NOTIFICATIONS_MASK = 0x400; |
| |
| public boolean needsRouting; |
| |
| protected String regionPath; |
| |
| protected int processorId; |
| |
| public DirectReplyProcessor processor; |
| |
| protected Object callbackArg; |
| |
| protected Operation op; |
| |
| public transient boolean directAck = false; |
| |
| protected transient DistributedCacheOperation owner; |
| |
| protected transient boolean appliedOperation = false; |
| |
| protected transient boolean closed = false; |
| |
| protected boolean hasOldValue; |
| protected boolean oldValueIsSerialized; |
| protected Object oldValue; |
| |
| protected boolean hasDelta = false; |
| |
| protected FilterRoutingInfo filterRouting; |
| |
| protected transient boolean sendDelta = true; |
| |
| /** concurrency versioning tag */ |
| protected VersionTag versionTag; |
| |
| protected transient short flags; |
| |
| protected boolean inhibitAllNotifications; |
| |
| public Operation getOperation() { |
| return this.op; |
| } |
| |
| /** sets the concurrency versioning tag for this message */ |
| public void setVersionTag(VersionTag tag) { |
| this.versionTag = tag; |
| } |
| |
| public VersionTag getVersionTag() { |
| return this.versionTag; |
| } |
| |
| @Override |
| public boolean isSevereAlertCompatible() { |
| // allow forced-disconnect processing for all cache op messages |
| return true; |
| } |
| |
| @Override |
| public DirectReplyProcessor getDirectReplyProcessor() { |
| return processor; |
| } |
| |
| @Override |
| public void registerProcessor() { |
| if (processor != null) { |
| this.processorId = this.processor.register(); |
| } |
| this.directAck = false; |
| } |
| |
| public void setFilterInfo(FilterRoutingInfo fInfo) { |
| this.filterRouting = fInfo; |
| } |
| |
| public void setInhibitNotificationsBit(boolean inhibit) { |
| this.inhibitAllNotifications = inhibit; |
| } |
| |
| public String getRegionPath() { |
| return this.regionPath; |
| } |
| |
| /** |
| * process a reply |
| * |
| * @return true if the reply-processor should continue to process this response |
| */ |
| boolean processReply(ReplyMessage reply, CacheOperationReplyProcessor processor) { |
| // notification that a reply has been received. Most messages |
| // don't do anything special here |
| return true; |
| } |
| |
| /** |
| * Add the cache event's old value to this message. We must propagate the old value when the |
| * receiver is doing GII and has listeners (CQs) that require the old value. |
| * |
| * @since GemFire 5.5 |
| * @param event the entry event that contains the old value |
| */ |
| public void appendOldValueToMessage(EntryEventImpl event) { |
| @Unretained |
| Object val = event.getRawOldValue(); |
| if (val == null || val == Token.NOT_AVAILABLE || val == Token.REMOVED_PHASE1 |
| || val == Token.REMOVED_PHASE2 || val == Token.DESTROYED || val == Token.TOMBSTONE) { |
| return; |
| } |
| event.exportOldValue(this); |
| } |
| |
| /** |
| * Insert this message's oldValue into the given event. This fixes bug 38382 by propagating old |
| * values with Entry level CacheOperationMessages during initial image transfer |
| * |
| * @since GemFire 5.5 |
| */ |
| public void setOldValueInEvent(EntryEventImpl event) { |
| CqService cqService = event.getRegion().getCache().getCqService(); |
| if (cqService.isRunning()/* || event.getOperation().guaranteesOldValue() */) { |
| event.setOldValueForQueryProcessing(); |
| if (!event.hasOldValue() && this.hasOldValue) { |
| if (this.oldValueIsSerialized) { |
| event.setSerializedOldValue((byte[]) this.oldValue); |
| } else { |
| event.setOldValue(this.oldValue); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Sets a flag in the message indicating that this message contains delta bytes. |
| * |
| * @since GemFire 6.1 |
| */ |
| protected void setHasDelta(boolean flag) { |
| this.hasDelta = flag; |
| } |
| |
| protected boolean hasDelta() { |
| return this.hasDelta; |
| } |
| |
| public FilterRoutingInfo getFilterInfo() { |
| return this.filterRouting; |
| } |
| |
| /** |
| * @since GemFire 4.2.3 |
| */ |
| protected transient boolean regionAllowsConflation; |
| |
| public boolean possibleDuplicate; |
| |
| @Override |
| public int getProcessorId() { |
| return this.processorId; |
| } |
| |
| @Override |
| public boolean containsRegionContentChange() { |
| return true; |
| } |
| |
| protected LocalRegion getLocalRegionForProcessing(ClusterDistributionManager dm) { |
| Assert.assertTrue(this.regionPath != null, "regionPath was null"); |
| InternalCache gfc = dm.getExistingCache(); |
| return (LocalRegion) gfc.getRegionByPathForProcessing(this.regionPath); |
| } |
| |
| @Override |
| protected void process(final ClusterDistributionManager dm) { |
| Throwable thr = null; |
| boolean sendReply = true; |
| |
| if (this.versionTag != null) { |
| this.versionTag.replaceNullIDs(getSender()); |
| } |
| |
| EntryLogger.setSource(this.getSender(), "p2p"); |
| final InitializationLevel oldLevel = |
| LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE); |
| try { |
| if (dm.getDMType() == ClusterDistributionManager.ADMIN_ONLY_DM_TYPE) { |
| // this was probably a multicast message |
| return; |
| } |
| |
| final LocalRegion lclRgn = getLocalRegionForProcessing(dm); |
| sendReply = false; |
| basicProcess(dm, lclRgn); |
| } catch (CancelException ignore) { |
| this.closed = true; |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} Cancelled: nothing to do", this); |
| } |
| } 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(); |
| thr = t; |
| } finally { |
| LocalRegion.setThreadInitLevelRequirement(oldLevel); |
| if (sendReply) { |
| ReplyException rex = null; |
| if (thr != null) { |
| rex = new ReplyException(thr); |
| } |
| sendReply(getSender(), processorId, rex, getReplySender(dm)); |
| } else if (thr != null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("In {}. process, got exception (NO_ACK)", getClass().getName(), thr); |
| } |
| } |
| EntryLogger.clearSource(); |
| } |
| } |
| |
| /** Return true if a reply should be sent */ |
| protected void basicProcess(ClusterDistributionManager dm, LocalRegion lclRgn) { |
| Throwable thr = null; |
| boolean sendReply = true; |
| |
| if (logger.isTraceEnabled()) { |
| logger.trace("DistributedCacheOperation.basicProcess: {}", this); |
| } |
| try { |
| // LocalRegion lclRgn = getRegionFromPath(dm.getSystem(), |
| // this.regionPath); |
| if (lclRgn == null) { |
| this.closed = true; |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} region not found, nothing to do", this); |
| } |
| return; |
| } |
| // Could this cause a deadlock, because this can block a P2P reader |
| // thread which might be needed to read the create region reply?? |
| // DAN - I don't think this does anything because process called |
| // LocalRegion.setThreadInitLevelRequirement |
| lclRgn.waitOnInitialization(); |
| // In some subclasses, lclRgn may be destroyed, so be careful not to |
| // allow a RegionDestroyedException to be thrown on lclRgn access |
| if (lclRgn.scope.isLocal()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} local scope region, nothing to do", this); |
| } |
| return; |
| } |
| DistributedRegion rgn = (DistributedRegion) lclRgn; |
| |
| // check to see if the region is in recovery mode, in which case |
| // we only operate if this is a DestroyRegion operation |
| if (rgn.getImageState().getInRecovery()) { |
| return; |
| } |
| |
| @Released |
| InternalCacheEvent event = createEvent(rgn); |
| try { |
| boolean isEntry = event.getOperation().isEntry(); |
| |
| if (isEntry && this.possibleDuplicate) { |
| ((EntryEventImpl) event).setPossibleDuplicate(true); |
| // If the state of the initial image yet to be received is unknown, |
| // we must not apply the event. It may already be reflected in the |
| // initial image state and, in fact, have been modified by subsequent |
| // events. This code path could be modified to pass the event to |
| // listeners and bridges, but it should not apply the change to the |
| // region |
| if (!rgn.isEventTrackerInitialized() |
| && (rgn.getDataPolicy().withReplication() || rgn.getDataPolicy().withPreloaded())) { |
| if (logger.isTraceEnabled()) { |
| logger.trace(LogMarker.DM_BRIDGE_SERVER_VERBOSE, |
| "Ignoring possible duplicate event"); |
| } |
| return; |
| } |
| } |
| |
| sendReply = operateOnRegion(event, dm) && sendReply; |
| } finally { |
| if (event instanceof EntryEventImpl) { |
| ((Releasable) event).release(); |
| } |
| } |
| } catch (RegionDestroyedException ignore) { |
| this.closed = true; |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} Region destroyed: nothing to do", this); |
| } |
| } catch (CancelException ignore) { |
| this.closed = true; |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} Cancelled: nothing to do", this); |
| } |
| } catch (DiskAccessException e) { |
| this.closed = true; |
| if (!lclRgn.isDestroyed()) { |
| logger.error("Got disk access exception, expected region to be destroyed", e); |
| } |
| } catch (EntryNotFoundException ignore) { |
| this.appliedOperation = true; |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} Entry not found, nothing to do", this); |
| } |
| } catch (InvalidDeltaException ide) { |
| ReplyException re = new ReplyException(ide); |
| sendReply = false; |
| sendReply(getSender(), processorId, re, getReplySender(dm)); |
| lclRgn.getCachePerfStats().incDeltaFullValuesRequested(); |
| } 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(); |
| thr = t; |
| } finally { |
| checkVersionIsRecorded(this.versionTag, lclRgn); |
| if (sendReply) { |
| ReplyException rex = null; |
| if (thr != null) { |
| rex = new ReplyException(thr); |
| } |
| sendReply(getSender(), processorId, rex, getReplySender(dm)); |
| } else if (thr != null) { |
| logger.error(String.format("Exception occurred while processing %s", |
| this), |
| thr); |
| } |
| } // finally |
| } |
| |
| public void sendReply(InternalDistributedMember recipient, int pId, ReplyException rex, |
| ReplySender dm) { |
| if (pId == 0 && (dm instanceof DistributionManager) && !this.directAck) {// Fix for #41871 |
| // distributed-no-ack message. Don't respond |
| } else { |
| ReplyMessage.send(recipient, pId, rex, dm, !this.appliedOperation, this.closed, false, |
| isInternal()); |
| } |
| } |
| |
| /** |
| * Ensure that a version tag has been recorded in the region's version vector. This makes note |
| * that the event has been received and processed but probably didn't affect the cache's state |
| * or it would have been recorded earlier. |
| * |
| * @param tag the version information |
| * @param r the affected region |
| */ |
| public void checkVersionIsRecorded(VersionTag tag, LocalRegion r) { |
| if (tag != null && !tag.isRecorded()) { // oops - someone forgot to record the event |
| if (r != null) { |
| RegionVersionVector v = r.getVersionVector(); |
| if (v != null) { |
| VersionSource mbr = tag.getMemberID(); |
| if (mbr == null) { |
| mbr = getSender(); |
| } |
| if (logger.isTraceEnabled()) { |
| logger.trace( |
| "recording version tag in RVV in basicProcess since it wasn't done earlier"); |
| } |
| v.recordVersion(mbr, tag); |
| } |
| } |
| } |
| } |
| |
| /** |
| * When an event is discarded because of an attempt to overwrite a more recent change we still |
| * need to deliver that event to clients. Clients can then perform their own concurrency checks |
| * on the event. |
| */ |
| protected void dispatchElidedEvent(LocalRegion rgn, EntryEventImpl ev) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("dispatching elided event: {}", ev); |
| } |
| ev.isConcurrencyConflict(true); |
| rgn.generateLocalFilterRouting(ev); |
| rgn.notifyBridgeClients(ev); |
| } |
| |
| protected abstract InternalCacheEvent createEvent(DistributedRegion rgn) |
| throws EntryNotFoundException; |
| |
| protected abstract boolean operateOnRegion(CacheEvent event, ClusterDistributionManager dm) |
| throws EntryNotFoundException; |
| |
| @Override |
| public String toString() { |
| StringBuilder buff = new StringBuilder(); |
| buff.append(getShortClassName()); |
| buff.append("(region path='"); // make sure this is the first one |
| buff.append(this.regionPath); |
| buff.append("'"); |
| appendFields(buff); |
| buff.append(")"); |
| return buff.toString(); |
| } |
| |
| // override and extend in subclasses |
| protected void appendFields(StringBuilder buff) { |
| buff.append("; sender="); |
| buff.append(getSender()); |
| buff.append("; callbackArg="); |
| buff.append(this.callbackArg); |
| buff.append("; processorId="); |
| buff.append(this.processorId); |
| buff.append("; op="); |
| buff.append(this.op); |
| buff.append("; applied="); |
| buff.append(this.appliedOperation); |
| buff.append("; directAck="); |
| buff.append(this.directAck); |
| buff.append("; posdup="); |
| buff.append(this.possibleDuplicate); |
| buff.append("; hasDelta="); |
| buff.append(this.hasDelta); |
| buff.append("; hasOldValue="); |
| buff.append(this.hasOldValue); |
| if (this.versionTag != null) { |
| buff.append("; version="); |
| buff.append(this.versionTag); |
| } |
| if (this.filterRouting != null) { |
| buff.append(" "); |
| buff.append(this.filterRouting.toString()); |
| } |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| short bits = in.readShort(); |
| short extBits = in.readShort(); |
| this.flags = bits; |
| setFlags(bits, in); |
| this.regionPath = DataSerializer.readString(in); |
| this.op = Operation.fromOrdinal(in.readByte()); |
| // TODO dirack There's really no reason to send this flag across the wire |
| // anymore |
| this.directAck = (bits & DIRECT_ACK_MASK) != 0; |
| this.possibleDuplicate = (bits & POSSIBLE_DUPLICATE_MASK) != 0; |
| if ((bits & CALLBACK_ARG_MASK) != 0) { |
| this.callbackArg = DataSerializer.readObject(in); |
| } |
| this.hasDelta = (bits & DELTA_MASK) != 0; |
| this.hasOldValue = (bits & OLD_VALUE_MASK) != 0; |
| if (this.hasOldValue) { |
| byte b = in.readByte(); |
| if (b == 0) { |
| this.oldValueIsSerialized = false; |
| } else if (b == 1) { |
| this.oldValueIsSerialized = true; |
| } else { |
| throw new IllegalStateException("expected 0 or 1"); |
| } |
| this.oldValue = DataSerializer.readByteArray(in); |
| } |
| boolean hasFilterInfo = (bits & FILTER_INFO_MASK) != 0; |
| this.needsRouting = (bits & NEEDS_ROUTING_MASK) != 0; |
| if (hasFilterInfo) { |
| this.filterRouting = new FilterRoutingInfo(); |
| InternalDataSerializer.invokeFromData(this.filterRouting, in); |
| } |
| if ((bits & VERSION_TAG_MASK) != 0) { |
| boolean persistentTag = (bits & PERSISTENT_TAG_MASK) != 0; |
| this.versionTag = VersionTag.create(persistentTag, in); |
| } |
| if ((extBits & INHIBIT_NOTIFICATIONS_MASK) != 0) { |
| this.inhibitAllNotifications = true; |
| } |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| short bits = 0; |
| short extendedBits = 0; |
| bits = computeCompressedShort(bits); |
| extendedBits = computeCompressedExtBits(extendedBits); |
| out.writeShort(bits); |
| out.writeShort(extendedBits); |
| if (this.processorId > 0) { |
| out.writeInt(this.processorId); |
| } |
| |
| DataSerializer.writeString(this.regionPath, out); |
| out.writeByte(this.op.ordinal); |
| if (this.callbackArg != null) { |
| DataSerializer.writeObject(this.callbackArg, out); |
| } |
| if (this.hasOldValue) { |
| out.writeByte(this.oldValueIsSerialized ? 1 : 0); |
| // the receiving side expects that the old value will have been serialized |
| // as a byte array |
| final byte policy = valueIsToDeserializationPolicy(this.oldValueIsSerialized); |
| final Object vObj; |
| final byte[] vBytes; |
| if (!this.oldValueIsSerialized && this.oldValue instanceof byte[]) { |
| vObj = null; |
| vBytes = (byte[]) this.oldValue; |
| } else { |
| vObj = this.oldValue; |
| vBytes = null; |
| } |
| writeValue(policy, vObj, vBytes, out); |
| } |
| if (this.filterRouting != null) { |
| InternalDataSerializer.invokeToData(this.filterRouting, out); |
| } |
| if (this.versionTag != null) { |
| InternalDataSerializer.invokeToData(this.versionTag, out); |
| } |
| } |
| |
| protected short computeCompressedShort(short bits) { |
| if (this.hasOldValue) { |
| bits |= OLD_VALUE_MASK; |
| } |
| if (this.directAck) { |
| bits |= DIRECT_ACK_MASK; |
| } |
| if (this.possibleDuplicate) { |
| bits |= POSSIBLE_DUPLICATE_MASK; |
| } |
| if (this.processorId != 0) { |
| bits |= HAS_PROCESSOR_ID; |
| } |
| if (this.callbackArg != null) { |
| bits |= CALLBACK_ARG_MASK; |
| } |
| if (this.hasDelta) { |
| bits |= DELTA_MASK; |
| } |
| if (this.filterRouting != null) { |
| bits |= FILTER_INFO_MASK; |
| } |
| if (this.needsRouting) { |
| bits |= NEEDS_ROUTING_MASK; |
| } |
| if (this.versionTag != null) { |
| bits |= VERSION_TAG_MASK; |
| } |
| if (this.versionTag instanceof DiskVersionTag) { |
| bits |= PERSISTENT_TAG_MASK; |
| } |
| if (inhibitAllNotifications) { |
| bits |= INHIBIT_NOTIFICATIONS_MASK; |
| } |
| return bits; |
| } |
| |
| protected short computeCompressedExtBits(short bits) { |
| if (inhibitAllNotifications) { |
| bits |= INHIBIT_NOTIFICATIONS_MASK; |
| } |
| return bits; |
| } |
| |
| protected void setFlags(short bits, DataInput in) throws IOException, ClassNotFoundException { |
| if ((bits & HAS_PROCESSOR_ID) != 0) { |
| this.processorId = in.readInt(); |
| ReplyProcessor21.setMessageRPId(this.processorId); |
| } |
| } |
| |
| @Override |
| public boolean supportsDirectAck() { |
| return this.directAck; |
| } |
| |
| public void setSendDelta(boolean sendDelta) { |
| this.sendDelta = sendDelta; |
| } |
| |
| @Override |
| public boolean prefersOldSerialized() { |
| return true; |
| } |
| |
| @Override |
| public boolean isUnretainedOldReferenceOk() { |
| return true; |
| } |
| |
| @Override |
| public boolean isCachedDeserializableValueOk() { |
| return false; |
| } |
| |
| @Override |
| public void importOldObject(Object ov, boolean isSerialized) { |
| this.oldValueIsSerialized = isSerialized; |
| this.oldValue = ov; |
| this.hasOldValue = true; |
| } |
| |
| @Override |
| public void importOldBytes(byte[] ov, boolean isSerialized) { |
| this.oldValueIsSerialized = isSerialized; |
| this.oldValue = ov; |
| this.hasOldValue = true; |
| } |
| |
| protected boolean notifiesSerialGatewaySender(ClusterDistributionManager dm) { |
| final InitializationLevel oldLevel = LocalRegion.setThreadInitLevelRequirement(ANY_INIT); |
| try { |
| LocalRegion lr = getLocalRegionForProcessing(dm); |
| if (lr == null) { |
| return false; |
| } |
| return lr.notifiesSerialGatewaySender(); |
| } catch (RuntimeException ignore) { |
| return false; |
| } finally { |
| LocalRegion.setThreadInitLevelRequirement(oldLevel); |
| } |
| } |
| } |
| |
| /** Custom subclass that keeps all ReplyExceptions */ |
| private static class ReliableCacheReplyProcessor extends CacheOperationReplyProcessor { |
| |
| private final Set failedMembers; |
| |
| private final DistributionManager dm; |
| |
| public ReliableCacheReplyProcessor(InternalDistributedSystem system, Collection initMembers, |
| Set departedMembers) { |
| super(system, initMembers); |
| this.dm = system.getDistributionManager(); |
| this.failedMembers = departedMembers; |
| } |
| |
| @Override |
| protected synchronized void processException(DistributionMessage dmsg, ReplyException ex) { |
| Throwable cause = ex.getCause(); |
| // only interested in CacheClosedException and RegionDestroyedException |
| if (cause instanceof CancelException || cause instanceof RegionDestroyedException) { |
| this.failedMembers.add(dmsg.getSender()); |
| } else { |
| // allow superclass to handle all other exceptions |
| this.failedMembers.add(dmsg.getSender()); |
| super.processException(dmsg, ex); |
| } |
| } |
| |
| @Override |
| protected void process(DistributionMessage dmsg, boolean warn) { |
| if (dmsg instanceof ReplyMessage && ((ReplyMessage) dmsg).getIgnored()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} replied with ignored true", dmsg.getSender()); |
| } |
| this.failedMembers.add(dmsg.getSender()); |
| } |
| super.process(dmsg, warn); |
| } |
| } |
| |
| static class CacheOperationReplyProcessor extends DirectReplyProcessor { |
| public CacheOperationMessage msg; |
| |
| public CopyOnWriteHashSet<InternalDistributedMember> closedMembers = new CopyOnWriteHashSet<>(); |
| |
| public CacheOperationReplyProcessor(InternalDistributedSystem system, Collection initMembers) { |
| super(system, initMembers); |
| } |
| |
| @Override |
| protected void process(final DistributionMessage dmsg, boolean warn) { |
| if (dmsg instanceof ReplyMessage) { |
| ReplyMessage replyMessage = (ReplyMessage) dmsg; |
| if (msg != null) { |
| boolean discard = !msg.processReply(replyMessage, this); |
| if (discard) { |
| return; |
| } |
| } |
| if (replyMessage.getClosed()) { |
| closedMembers.add(replyMessage.getSender()); |
| } |
| } |
| |
| super.process(dmsg, warn); |
| } |
| |
| } |
| } |
