| /* |
| * 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.util.Arrays; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.cache.CacheEvent; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.RegionAttributes; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.persistence.PersistentReplicatesOfflineException; |
| import org.apache.geode.cache.query.internal.cq.CqService; |
| import org.apache.geode.distributed.internal.DirectReplyProcessor; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.ByteArrayDataInput; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.Version; |
| import org.apache.geode.internal.cache.DistributedPutAllOperation.EntryVersionsList; |
| import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo; |
| import org.apache.geode.internal.cache.ha.ThreadIdentifier; |
| import org.apache.geode.internal.cache.partitioned.PutAllPRMessage; |
| import org.apache.geode.internal.cache.partitioned.RemoveAllPRMessage; |
| import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID; |
| import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList; |
| import org.apache.geode.internal.cache.tx.RemotePutAllMessage; |
| import org.apache.geode.internal.cache.tx.RemoteRemoveAllMessage; |
| import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException; |
| import org.apache.geode.internal.cache.versions.DiskVersionTag; |
| 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.offheap.annotations.Released; |
| import org.apache.geode.internal.offheap.annotations.Retained; |
| import org.apache.geode.internal.offheap.annotations.Unretained; |
| |
| /** |
| * Handles distribution of a Region.removeAll operation. |
| * |
| * TODO: extend DistributedCacheOperation instead of AbstractUpdateOperation |
| * |
| * @since GemFire 8.1 |
| */ |
| public class DistributedRemoveAllOperation extends AbstractUpdateOperation { |
| private static final Logger logger = LogService.getLogger(); |
| |
| /** |
| * Release is called by freeOffHeapResources. |
| */ |
| @Retained |
| protected final RemoveAllEntryData[] removeAllData; |
| |
| public int removeAllDataSize; |
| |
| protected boolean isBridgeOp = false; |
| |
| static final byte USED_FAKE_EVENT_ID = 0x01; |
| static final byte NOTIFY_ONLY = 0x02; |
| static final byte FILTER_ROUTING = 0x04; |
| static final byte VERSION_TAG = 0x08; |
| static final byte POSDUP = 0x10; |
| static final byte PERSISTENT_TAG = 0x20; |
| static final byte HAS_CALLBACKARG = 0x40; |
| static final byte HAS_TAILKEY = (byte) 0x80; |
| |
| public DistributedRemoveAllOperation(CacheEvent event, int size, boolean isBridgeOp) { |
| super(event, ((EntryEventImpl) event).getEventTime(0L)); |
| this.removeAllData = new RemoveAllEntryData[size]; |
| this.removeAllDataSize = 0; |
| this.isBridgeOp = isBridgeOp; |
| } |
| |
| /** |
| * return if the operation is bridge operation |
| */ |
| public boolean isBridgeOperation() { |
| return this.isBridgeOp; |
| } |
| |
| public RemoveAllEntryData[] getRemoveAllEntryData() { |
| return removeAllData; |
| } |
| |
| public void setRemoveAllEntryData(RemoveAllEntryData[] removeAllEntryData) { |
| for (int i = 0; i < removeAllEntryData.length; i++) { |
| removeAllData[i] = removeAllEntryData[i]; |
| } |
| this.removeAllDataSize = removeAllEntryData.length; |
| } |
| |
| /** |
| * Add an entry that this removeAll operation should distribute. |
| */ |
| public void addEntry(RemoveAllEntryData removeAllEntry) { |
| this.removeAllData[this.removeAllDataSize] = removeAllEntry; |
| this.removeAllDataSize += 1; |
| } |
| |
| /** |
| * Add an entry that this removeAll operation should distribute. |
| */ |
| public void addEntry(EntryEventImpl ev) { |
| this.removeAllData[this.removeAllDataSize] = new RemoveAllEntryData(ev); |
| this.removeAllDataSize += 1; |
| } |
| |
| /** |
| * Add an entry that this removeAll operation should distribute. This method is for a special |
| * case: the callback will be called after this in hasSeenEvent() case, so we should change the |
| * status beforehand |
| */ |
| public void addEntry(EntryEventImpl ev, boolean newCallbackInvoked) { |
| this.removeAllData[this.removeAllDataSize] = new RemoveAllEntryData(ev); |
| this.removeAllData[this.removeAllDataSize].setCallbacksInvoked(newCallbackInvoked); |
| this.removeAllDataSize += 1; |
| } |
| |
| /** |
| * Add an entry for PR bucket's msg. |
| * |
| * @param ev event to be added |
| * @param bucketId message is for this bucket |
| */ |
| public void addEntry(EntryEventImpl ev, Integer bucketId) { |
| this.removeAllData[this.removeAllDataSize] = new RemoveAllEntryData(ev); |
| this.removeAllData[this.removeAllDataSize].setBucketId(bucketId); |
| this.removeAllDataSize += 1; |
| } |
| |
| /** |
| * set using fake thread id |
| * |
| * @param status whether the entry is using fake event id |
| */ |
| public void setUseFakeEventId(boolean status) { |
| for (int i = 0; i < removeAllDataSize; i++) { |
| removeAllData[i].setUsedFakeEventId(status); |
| } |
| } |
| |
| /** |
| * In the originating cache, this returns an iterator on the list of events caused by the |
| * removeAll operation. This is cached for listener notification purposes. The iterator is |
| * guaranteed to return events in the order they are present in putAllData[] |
| */ |
| public Iterator eventIterator() { |
| return new Iterator() { |
| int position = 0; |
| |
| @Override |
| public boolean hasNext() { |
| return DistributedRemoveAllOperation.this.removeAllDataSize > position; |
| }; |
| |
| @Override |
| @Unretained |
| public Object next() { |
| @Unretained |
| EntryEventImpl ev = getEventForPosition(position); |
| position++; |
| return ev; |
| }; |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException(); |
| }; |
| }; |
| } |
| |
| public void freeOffHeapResources() { |
| // I do not use eventIterator here because it forces the lazy creation of EntryEventImpl by |
| // calling getEventForPosition. |
| for (int i = 0; i < this.removeAllDataSize; i++) { |
| RemoveAllEntryData entry = this.removeAllData[i]; |
| if (entry != null && entry.event != null) { |
| entry.event.release(); |
| } |
| } |
| } |
| |
| @Unretained |
| public EntryEventImpl getEventForPosition(int position) { |
| RemoveAllEntryData entry = this.removeAllData[position]; |
| if (entry == null) { |
| return null; |
| } |
| if (entry.event != null) { |
| return entry.event; |
| } |
| LocalRegion region = (LocalRegion) this.event.getRegion(); |
| // owned by this.removeAllData once entry.event = ev is done |
| @Retained |
| EntryEventImpl ev = EntryEventImpl.create(region, entry.getOp(), entry.getKey(), |
| null/* value */, this.event.getCallbackArgument(), false /* originRemote */, |
| this.event.getDistributedMember(), this.event.isGenerateCallbacks(), entry.getEventID()); |
| boolean returnedEv = false; |
| try { |
| ev.setPossibleDuplicate(entry.isPossibleDuplicate()); |
| ev.setIsRedestroyedEntry(entry.getRedestroyedEntry()); |
| if (entry.versionTag != null && region.getConcurrencyChecksEnabled()) { |
| VersionSource id = entry.versionTag.getMemberID(); |
| if (id != null) { |
| entry.versionTag.setMemberID(ev.getRegion().getVersionVector().getCanonicalId(id)); |
| } |
| ev.setVersionTag(entry.versionTag); |
| } |
| |
| entry.event = ev; |
| returnedEv = true; |
| ev.setOldValue(entry.getOldValue()); |
| CqService cqService = region.getCache().getCqService(); |
| if (cqService.isRunning() && !entry.getOp().isCreate() && !ev.hasOldValue()) { |
| ev.setOldValueForQueryProcessing(); |
| } |
| ev.setInvokePRCallbacks(!entry.isNotifyOnly()); |
| if (getBaseEvent().getContext() != null) { |
| ev.setContext(getBaseEvent().getContext()); |
| } |
| ev.callbacksInvoked(entry.isCallbacksInvoked()); |
| ev.setTailKey(entry.getTailKey()); |
| return ev; |
| } finally { |
| if (!returnedEv) { |
| ev.release(); |
| } |
| } |
| } |
| |
| public EntryEventImpl getBaseEvent() { |
| return getEvent(); |
| } |
| |
| /** |
| * Data that represents a single entry being RemoveAll'd. |
| */ |
| public static class RemoveAllEntryData { |
| |
| final Object key; |
| |
| private final Object oldValue; |
| |
| private final Operation op; |
| |
| private EventID eventID; |
| |
| transient EntryEventImpl event; |
| |
| private Integer bucketId = Integer.valueOf(-1); |
| |
| protected transient boolean callbacksInvoked = false; |
| |
| public FilterRoutingInfo filterRouting; |
| |
| // One flags byte for all booleans |
| protected byte flags = 0x00; |
| |
| // TODO: Yogesh, this should be intialized and sent on wire only when |
| // parallel wan is enabled |
| private Long tailKey = 0L; |
| |
| public VersionTag versionTag; |
| |
| transient boolean inhibitDistribution; |
| |
| transient boolean redestroyedEntry; |
| |
| /** |
| * Constructor to use when preparing to send putall data out |
| */ |
| public RemoveAllEntryData(EntryEventImpl event) { |
| |
| this.key = event.getKey(); |
| Object oldValue = event.getRawOldValue(); |
| |
| if (oldValue == Token.NOT_AVAILABLE || Token.isRemoved(oldValue)) { |
| this.oldValue = null; |
| } else { |
| this.oldValue = oldValue; |
| } |
| |
| this.op = event.getOperation(); |
| this.eventID = event.getEventId(); |
| this.tailKey = event.getTailKey(); |
| this.versionTag = event.getVersionTag(); |
| |
| setNotifyOnly(!event.getInvokePRCallbacks()); |
| setCallbacksInvoked(event.callbacksInvoked()); |
| setPossibleDuplicate(event.isPossibleDuplicate()); |
| setInhibitDistribution(event.getInhibitDistribution()); |
| setRedestroyedEntry(event.getIsRedestroyedEntry()); |
| } |
| |
| /** |
| * Constructor to use when receiving a putall from someone else |
| */ |
| public RemoveAllEntryData(DataInput in, EventID baseEventID, int idx, Version version, |
| ByteArrayDataInput bytesIn) throws IOException, ClassNotFoundException { |
| this.key = DataSerializer.readObject(in); |
| this.oldValue = null; |
| this.op = Operation.fromOrdinal(in.readByte()); |
| this.flags = in.readByte(); |
| if ((this.flags & FILTER_ROUTING) != 0) { |
| this.filterRouting = (FilterRoutingInfo) DataSerializer.readObject(in); |
| } |
| if ((this.flags & VERSION_TAG) != 0) { |
| boolean persistentTag = (this.flags & PERSISTENT_TAG) != 0; |
| this.versionTag = VersionTag.create(persistentTag, in); |
| } |
| if (isUsedFakeEventId()) { |
| this.eventID = new EventID(); |
| InternalDataSerializer.invokeFromData(this.eventID, in); |
| } else { |
| this.eventID = new EventID(baseEventID, idx); |
| } |
| if ((this.flags & HAS_TAILKEY) != 0) { |
| this.tailKey = DataSerializer.readLong(in); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder sb = new StringBuilder(50); |
| sb.append("(").append(getKey()).append(",").append(getOldValue()); |
| if (this.bucketId > 0) { |
| sb.append(", b").append(this.bucketId); |
| } |
| if (versionTag != null) { |
| sb.append(",v").append(versionTag.getEntryVersion()) |
| .append(",rv=" + versionTag.getRegionVersion()); |
| } |
| if (filterRouting != null) { |
| sb.append(", ").append(filterRouting); |
| } |
| sb.append(")"); |
| return sb.toString(); |
| } |
| |
| void setSender(InternalDistributedMember sender) { |
| if (this.versionTag != null) { |
| this.versionTag.replaceNullIDs(sender); |
| } |
| } |
| |
| /** |
| * Used to serialize this instances data to <code>out</code>. If changes are made to this method |
| * make sure that it is backwards compatible by creating toDataPreXX methods. Also make sure |
| * that the callers to this method are backwards compatible by creating toDataPreXX methods for |
| * them even if they are not changed. <br> |
| * Callers for this method are: <br> |
| * {@link RemoveAllMessage#toData(DataOutput)} <br> |
| * {@link PutAllPRMessage#toData(DataOutput)} <br> |
| * {@link RemotePutAllMessage#toData(DataOutput)} <br> |
| */ |
| public void toData(final DataOutput out) throws IOException { |
| Object key = this.key; |
| DataSerializer.writeObject(key, out); |
| |
| out.writeByte(this.op.ordinal); |
| byte bits = this.flags; |
| if (this.filterRouting != null) |
| bits |= FILTER_ROUTING; |
| if (this.versionTag != null) { |
| bits |= VERSION_TAG; |
| if (this.versionTag instanceof DiskVersionTag) { |
| bits |= PERSISTENT_TAG; |
| } |
| } |
| // TODO: Yogesh, this should be conditional, |
| // make sure that we sent it on wire only |
| // when parallel wan is enabled |
| bits |= HAS_TAILKEY; |
| out.writeByte(bits); |
| |
| if (this.filterRouting != null) { |
| DataSerializer.writeObject(this.filterRouting, out); |
| } |
| if (this.versionTag != null) { |
| InternalDataSerializer.invokeToData(this.versionTag, out); |
| } |
| if (isUsedFakeEventId()) { |
| // fake event id should be serialized |
| InternalDataSerializer.invokeToData(this.eventID, out); |
| } |
| // TODO: Yogesh, this should be conditional, |
| // make sure that we sent it on wire only |
| // when parallel wan is enabled |
| DataSerializer.writeLong(this.tailKey, out); |
| } |
| |
| /** |
| * Returns the key |
| */ |
| public Object getKey() { |
| return this.key; |
| } |
| |
| /** |
| * Returns the old value |
| */ |
| public Object getOldValue() { |
| return this.oldValue; |
| } |
| |
| public Long getTailKey() { |
| return this.tailKey; |
| } |
| |
| public void setTailKey(Long key) { |
| this.tailKey = key; |
| } |
| |
| /** |
| * Returns the operation |
| */ |
| public Operation getOp() { |
| return this.op; |
| } |
| |
| public EventID getEventID() { |
| return this.eventID; |
| } |
| |
| /** |
| * change event id for the entry |
| * |
| * @param eventId new event id |
| */ |
| public void setEventId(EventID eventId) { |
| this.eventID = eventId; |
| } |
| |
| /** |
| * change bucket id for the entry |
| * |
| * @param bucketId new bucket id |
| */ |
| public void setBucketId(Integer bucketId) { |
| this.bucketId = bucketId; |
| } |
| |
| /** |
| * get bucket id for the entry |
| * |
| * @return bucket id |
| */ |
| public Integer getBucketId() { |
| return this.bucketId; |
| } |
| |
| /** |
| * change event id into fake event id The algorithm is to change the threadid into |
| * bucketid*MAX_THREAD_PER_CLIENT+oldthreadid. So from the log, we can derive the original |
| * thread id. |
| * |
| * @return wether current event id is fake or not new bucket id |
| */ |
| public boolean setFakeEventID() { |
| if (bucketId.intValue() < 0) { |
| return false; |
| } |
| |
| if (!isUsedFakeEventId()) { |
| // assign a fake big thread id. bucket id starts from 0. In order to distinguish |
| // with other read thread, let bucket id starts from 1 in fake thread id |
| long threadId = ThreadIdentifier.createFakeThreadIDForBulkOp(bucketId.intValue(), |
| eventID.getThreadID()); |
| this.eventID = new EventID(eventID.getMembershipID(), threadId, eventID.getSequenceID()); |
| this.setUsedFakeEventId(true); |
| } |
| return true; |
| } |
| |
| public boolean isUsedFakeEventId() { |
| return (flags & USED_FAKE_EVENT_ID) != 0; |
| } |
| |
| public void setUsedFakeEventId(boolean usedFakeEventId) { |
| if (usedFakeEventId) { |
| flags |= USED_FAKE_EVENT_ID; |
| } else { |
| flags &= ~(USED_FAKE_EVENT_ID); |
| } |
| } |
| |
| public boolean isNotifyOnly() { |
| return (flags & NOTIFY_ONLY) != 0; |
| } |
| |
| public void setNotifyOnly(boolean notifyOnly) { |
| if (notifyOnly) { |
| flags |= NOTIFY_ONLY; |
| } else { |
| flags &= ~(NOTIFY_ONLY); |
| } |
| } |
| |
| boolean isPossibleDuplicate() { |
| return (this.flags & POSDUP) != 0; |
| } |
| |
| public void setPossibleDuplicate(boolean possibleDuplicate) { |
| if (possibleDuplicate) { |
| flags |= POSDUP; |
| } else { |
| flags &= ~(POSDUP); |
| } |
| } |
| |
| public boolean isInhibitDistribution() { |
| return this.inhibitDistribution; |
| } |
| |
| public void setInhibitDistribution(boolean inhibitDistribution) { |
| this.inhibitDistribution = inhibitDistribution; |
| } |
| |
| public boolean getRedestroyedEntry() { |
| return redestroyedEntry; |
| } |
| |
| public void setRedestroyedEntry(boolean redestroyedEntry) { |
| this.redestroyedEntry = redestroyedEntry; |
| } |
| |
| public boolean isCallbacksInvoked() { |
| return this.callbacksInvoked; |
| } |
| |
| public void setCallbacksInvoked(boolean callbacksInvoked) { |
| this.callbacksInvoked = callbacksInvoked; |
| } |
| } |
| |
| @Override |
| protected FilterRoutingInfo getRecipientFilterRouting(Set cacheOpRecipients) { |
| // for removeAll, we need to determine the routing information for each event and |
| // create a consolidated routing object representing all events that can be |
| // used for distribution |
| CacheDistributionAdvisor advisor; |
| LocalRegion region = (LocalRegion) this.event.getRegion(); |
| if (region instanceof PartitionedRegion) { |
| advisor = ((PartitionedRegion) region).getCacheDistributionAdvisor(); |
| } else if (region.isUsedForPartitionedRegionBucket()) { |
| advisor = ((BucketRegion) region).getPartitionedRegion().getCacheDistributionAdvisor(); |
| } else { |
| advisor = ((DistributedRegion) region).getCacheDistributionAdvisor(); |
| } |
| FilterRoutingInfo consolidated = new FilterRoutingInfo(); |
| for (int i = 0; i < this.removeAllData.length; i++) { |
| @Unretained |
| EntryEventImpl ev = getEventForPosition(i); |
| if (ev != null) { |
| FilterRoutingInfo eventRouting = advisor.adviseFilterRouting(ev, cacheOpRecipients); |
| if (eventRouting != null) { |
| consolidated.addFilterInfo(eventRouting); |
| } |
| removeAllData[i].filterRouting = eventRouting; |
| } |
| } |
| // we need to create routing information for each PUT event |
| return consolidated; |
| } |
| |
| |
| @Override |
| protected FilterInfo getLocalFilterRouting(FilterRoutingInfo frInfo) { |
| FilterProfile fp = getRegion().getFilterProfile(); |
| if (fp == null) { |
| return null; |
| } |
| |
| // this will set the local FilterInfo in the events |
| if (this.removeAllData != null && this.removeAllData.length > 0) { |
| fp.getLocalFilterRoutingForRemoveAllOp(this, this.removeAllData); |
| } |
| |
| return null; |
| } |
| |
| @Override |
| protected CacheOperationMessage createMessage() { |
| EntryEventImpl event = getBaseEvent(); |
| RemoveAllMessage msg = new RemoveAllMessage(); |
| msg.eventId = event.getEventId(); |
| msg.context = event.getContext(); |
| return msg; |
| } |
| |
| /** |
| * Create RemoveAllPRMessage for notify only (to adjunct nodes) |
| * |
| * @param bucketId create message to send to this bucket |
| */ |
| public RemoveAllPRMessage createPRMessagesNotifyOnly(int bucketId) { |
| final EntryEventImpl event = getBaseEvent(); |
| RemoveAllPRMessage prMsg = new RemoveAllPRMessage(bucketId, removeAllDataSize, true, |
| event.isPossibleDuplicate(), !event.isGenerateCallbacks(), event.getCallbackArgument()); |
| if (event.getContext() != null) { |
| prMsg.setBridgeContext(event.getContext()); |
| } |
| |
| // will not recover event id here |
| for (int i = 0; i < removeAllDataSize; i++) { |
| prMsg.addEntry(removeAllData[i]); |
| } |
| |
| return prMsg; |
| } |
| |
| /** |
| * Create RemoveAllPRMessages for primary buckets out of this op |
| * |
| * @return a HashMap contain RemoveAllPRMessages, key is bucket id |
| */ |
| public HashMap<Integer, RemoveAllPRMessage> createPRMessages() { |
| // getFilterRecipients(Collections.EMPTY_SET); // establish filter recipient routing information |
| HashMap<Integer, RemoveAllPRMessage> prMsgMap = new HashMap<Integer, RemoveAllPRMessage>(); |
| final EntryEventImpl event = getBaseEvent(); |
| |
| for (int i = 0; i < removeAllDataSize; i++) { |
| Integer bucketId = removeAllData[i].getBucketId(); |
| RemoveAllPRMessage prMsg = prMsgMap.get(bucketId); |
| if (prMsg == null) { |
| prMsg = new RemoveAllPRMessage(bucketId.intValue(), removeAllDataSize, false, |
| event.isPossibleDuplicate(), !event.isGenerateCallbacks(), event.getCallbackArgument()); |
| prMsg |
| .setTransactionDistributed(event.getRegion().getCache().getTxManager().isDistributed()); |
| |
| // set dpao's context(original sender) into each PutAllMsg |
| // dpao's event's context could be null if it's P2P putAll in PR |
| if (event.getContext() != null) { |
| prMsg.setBridgeContext(event.getContext()); |
| } |
| } |
| |
| // Modify the event id, assign new thread id and new sequence id |
| // We have to set fake event id here, because we cannot derive old event id from baseId+idx as |
| // we |
| // did in DR's PutAllMessage. |
| removeAllData[i].setFakeEventID(); |
| // we only save the reference in prMsg. No duplicate copy |
| prMsg.addEntry(removeAllData[i]); |
| prMsgMap.put(bucketId, prMsg); |
| } |
| return prMsgMap; |
| } |
| |
| @Override |
| protected void initMessage(CacheOperationMessage msg, DirectReplyProcessor proc) { |
| super.initMessage(msg, proc); |
| RemoveAllMessage m = (RemoveAllMessage) msg; |
| |
| // if concurrency checks are enabled and this is not a replicated |
| // region we need to see if any of the entries have no versions and, |
| // if so, cull them out and send a 1-hop message to a replicate that |
| // can generate a version for the operation |
| |
| RegionAttributes attr = this.event.getRegion().getAttributes(); |
| if (attr.getConcurrencyChecksEnabled() && !attr.getDataPolicy().withReplication() |
| && attr.getScope() != Scope.GLOBAL) { |
| if (attr.getDataPolicy() == DataPolicy.EMPTY) { |
| // all entries are without version tags |
| boolean success = RemoteRemoveAllMessage.distribute((EntryEventImpl) this.event, |
| this.removeAllData, this.removeAllDataSize); |
| if (success) { |
| m.callbackArg = this.event.getCallbackArgument(); |
| m.removeAllData = new RemoveAllEntryData[0]; |
| m.removeAllDataSize = 0; |
| m.skipCallbacks = !event.isGenerateCallbacks(); |
| |
| return; |
| |
| } else if (!getRegion().getGenerateVersionTag()) { |
| // Fix for #45934. We can't continue if we need versions and we failed |
| // to distribute versionless entries. |
| throw new PersistentReplicatesOfflineException(); |
| } |
| } else { |
| // some entries may have Create ops - these will not have version tags |
| RemoveAllEntryData[] versionless = selectVersionlessEntries(); |
| if (logger.isTraceEnabled()) { |
| logger.trace("Found these versionless entries: {}", Arrays.toString(versionless)); |
| } |
| if (versionless.length > 0) { |
| boolean success = RemoteRemoveAllMessage.distribute((EntryEventImpl) this.event, |
| versionless, versionless.length); |
| if (success) { |
| versionless = null; |
| RemoveAllEntryData[] versioned = selectVersionedEntries(); |
| if (logger.isTraceEnabled()) { |
| logger.trace("Found these remaining versioned entries: {}", |
| Arrays.toString(versioned)); |
| } |
| m.callbackArg = this.event.getCallbackArgument(); |
| m.removeAllData = versioned; |
| m.removeAllDataSize = versioned.length; |
| m.skipCallbacks = !event.isGenerateCallbacks(); |
| return; |
| |
| } else if (!getRegion().getGenerateVersionTag()) { |
| // Fix for #45934. We can't continue if we need versions and we failed |
| // to distribute versionless entries. |
| throw new PersistentReplicatesOfflineException(); |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug("All entries have versions, so using normal DPAO message"); |
| } |
| } |
| } |
| } |
| m.callbackArg = this.event.getCallbackArgument(); |
| m.removeAllData = this.removeAllData; |
| m.removeAllDataSize = this.removeAllDataSize; |
| m.skipCallbacks = !event.isGenerateCallbacks(); |
| } |
| |
| |
| @Override |
| protected boolean shouldAck() { |
| // bug #45704 - RemotePutAllOp's DPAO in another server conflicts with lingering DPAO from same |
| // thread, so |
| // we require an ACK if concurrency checks are enabled to make sure that the previous op has |
| // finished first. |
| return super.shouldAck() || getRegion().getConcurrencyChecksEnabled(); |
| } |
| |
| private RemoveAllEntryData[] selectVersionlessEntries() { |
| int resultSize = this.removeAllData.length; |
| for (int i = 0; i < this.removeAllData.length; i++) { |
| RemoveAllEntryData p = this.removeAllData[i]; |
| if (p == null || p.isInhibitDistribution()) { |
| resultSize--; |
| } else if (p.versionTag != null && p.versionTag.hasValidVersion()) { |
| resultSize--; |
| } |
| } |
| RemoveAllEntryData[] result = new RemoveAllEntryData[resultSize]; |
| int ri = 0; |
| for (int i = 0; i < this.removeAllData.length; i++) { |
| RemoveAllEntryData p = this.removeAllData[i]; |
| if (p == null || p.isInhibitDistribution()) { |
| continue; // skip blanks |
| } else if (p.versionTag != null && p.versionTag.hasValidVersion()) { |
| continue; // skip versioned |
| } |
| // what remains is versionless |
| result[ri++] = p; |
| } |
| return result; |
| } |
| |
| private RemoveAllEntryData[] selectVersionedEntries() { |
| int resultSize = 0; |
| for (int i = 0; i < this.removeAllData.length; i++) { |
| RemoveAllEntryData p = this.removeAllData[i]; |
| if (p == null || p.isInhibitDistribution()) { |
| continue; // skip blanks |
| } else if (p.versionTag != null && p.versionTag.hasValidVersion()) { |
| resultSize++; |
| } |
| } |
| RemoveAllEntryData[] result = new RemoveAllEntryData[resultSize]; |
| int ri = 0; |
| for (int i = 0; i < this.removeAllData.length; i++) { |
| RemoveAllEntryData p = this.removeAllData[i]; |
| if (p == null || p.isInhibitDistribution()) { |
| continue; // skip blanks |
| } else if (p.versionTag != null && p.versionTag.hasValidVersion()) { |
| result[ri++] = p; |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * version tags are gathered from local operations and remote operation responses. This method |
| * gathers all of them and stores them in the given list. |
| * |
| */ |
| protected void fillVersionedObjectList(VersionedObjectList list) { |
| for (RemoveAllEntryData entry : this.removeAllData) { |
| if (entry.versionTag != null) { |
| list.addKeyAndVersion(entry.key, entry.versionTag); |
| } |
| } |
| } |
| |
| |
| public static class RemoveAllMessage extends AbstractUpdateMessage // TODO extend |
| // CacheOperationMessage |
| // instead |
| { |
| |
| protected RemoveAllEntryData[] removeAllData; |
| |
| protected int removeAllDataSize; |
| |
| protected transient ClientProxyMembershipID context; |
| |
| protected boolean skipCallbacks; |
| |
| protected EventID eventId = null; |
| |
| protected static final short HAS_BRIDGE_CONTEXT = UNRESERVED_FLAGS_START; |
| protected static final short SKIP_CALLBACKS = (short) (HAS_BRIDGE_CONTEXT << 1); |
| |
| /** test to see if this message holds any data */ |
| public boolean isEmpty() { |
| return this.removeAllData.length == 0; |
| } |
| |
| /** |
| * Note this this is a "dummy" event since this message contains a list of entries each one of |
| * which has its own event. The key thing needed in this event is the region. This is the event |
| * that gets passed to basicOperateOnRegion |
| */ |
| @Override |
| @Retained |
| protected InternalCacheEvent createEvent(DistributedRegion rgn) throws EntryNotFoundException { |
| // Gester: We have to specify eventId for the message of MAP |
| @Retained |
| EntryEventImpl event = EntryEventImpl.create(rgn, Operation.REMOVEALL_DESTROY, null /* key */, |
| null/* value */, this.callbackArg, true /* originRemote */, getSender()); |
| if (this.context != null) { |
| event.context = this.context; |
| } |
| event.setPossibleDuplicate(this.possibleDuplicate); |
| event.setEventId(this.eventId); |
| return event; |
| } |
| |
| @Override |
| public void appendFields(StringBuilder sb) { |
| super.appendFields(sb); |
| if (eventId != null) { |
| sb.append("; eventId=").append(this.eventId); |
| } |
| sb.append("; entries=").append(this.removeAllDataSize); |
| if (removeAllDataSize <= 20) { |
| // 20 is a size for test |
| sb.append("; entry values=").append(Arrays.toString(this.removeAllData)); |
| } |
| } |
| |
| /** |
| * Does the "remove" of one entry for a "removeAll" operation. Note it calls back to |
| * AbstractUpdateOperation.UpdateMessage#basicOperationOnRegion |
| * |
| * @param entry the entry being removed |
| * @param rgn the region the entry is removed from |
| */ |
| public void doEntryRemove(RemoveAllEntryData entry, DistributedRegion rgn) { |
| @Released |
| EntryEventImpl ev = RemoveAllMessage.createEntryEvent(entry, getSender(), this.context, rgn, |
| this.possibleDuplicate, this.needsRouting, this.callbackArg, true, skipCallbacks); |
| // rgn.getLogWriterI18n().info(String.format("%s", "RemoveAllMessage.doEntryRemove |
| // sender=" + getSender() + |
| // " event="+ev)); |
| // we don't need to set old value here, because the msg is from remote. local old value will |
| // get from next step |
| try { |
| if (ev.getVersionTag() != null) { |
| checkVersionTag(rgn, ev.getVersionTag()); |
| } |
| // TODO check all removeAll basicDestroy calls done on the farside and make sure |
| // "cacheWrite" is false |
| rgn.basicDestroy(ev, false, null); |
| } catch (EntryNotFoundException ignore) { |
| this.appliedOperation = true; |
| } catch (ConcurrentCacheModificationException e) { |
| dispatchElidedEvent(rgn, ev); |
| this.appliedOperation = false; |
| } finally { |
| if (ev.hasValidVersionTag() && !ev.getVersionTag().isRecorded()) { |
| if (rgn.getVersionVector() != null) { |
| rgn.getVersionVector().recordVersion(getSender(), ev.getVersionTag()); |
| } |
| } |
| ev.release(); |
| } |
| } |
| |
| /** |
| * create an event for a RemoveAllEntryData element |
| * |
| * @return the event to be used in applying the element |
| */ |
| @Retained |
| public static EntryEventImpl createEntryEvent(RemoveAllEntryData entry, |
| InternalDistributedMember sender, ClientProxyMembershipID context, DistributedRegion rgn, |
| boolean possibleDuplicate, boolean needsRouting, Object callbackArg, boolean originRemote, |
| boolean skipCallbacks) { |
| final Object key = entry.getKey(); |
| EventID evId = entry.getEventID(); |
| @Retained |
| EntryEventImpl ev = EntryEventImpl.create(rgn, entry.getOp(), key, null/* value */, |
| callbackArg, originRemote, sender, !skipCallbacks, evId); |
| boolean returnedEv = false; |
| try { |
| if (context != null) { |
| ev.context = context; |
| } |
| ev.setPossibleDuplicate(possibleDuplicate); |
| ev.setVersionTag(entry.versionTag); |
| // if (needsRouting) { |
| // FilterProfile fp = rgn.getFilterProfile(); |
| // if (fp != null) { |
| // FilterInfo fi = fp.getLocalFilterRouting(ev); |
| // ev.setLocalFilterInfo(fi); |
| // } |
| // } |
| if (entry.filterRouting != null) { |
| InternalDistributedMember id = rgn.getMyId(); |
| ev.setLocalFilterInfo(entry.filterRouting.getFilterInfo(id)); |
| } |
| /** |
| * Setting tailKey for the secondary bucket here. Tail key was update by the primary. |
| */ |
| ev.setTailKey(entry.getTailKey()); |
| returnedEv = true; |
| return ev; |
| } finally { |
| if (!returnedEv) { |
| ev.release(); |
| } |
| } |
| } |
| |
| @Override |
| protected void basicOperateOnRegion(EntryEventImpl ev, final DistributedRegion rgn) { |
| for (int i = 0; i < removeAllDataSize; ++i) { |
| if (removeAllData[i].versionTag != null) { |
| checkVersionTag(rgn, removeAllData[i].versionTag); |
| } |
| } |
| |
| rgn.syncBulkOp(new Runnable() { |
| @Override |
| public void run() { |
| for (int i = 0; i < removeAllDataSize; ++i) { |
| if (logger.isTraceEnabled()) { |
| logger.trace("removeAll processing {} with {}", removeAllData[i], |
| removeAllData[i].versionTag); |
| } |
| removeAllData[i].setSender(sender); |
| doEntryRemove(removeAllData[i], rgn); |
| } |
| } |
| }, ev.getEventId()); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return REMOVE_ALL_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| |
| super.fromData(in); |
| this.eventId = (EventID) DataSerializer.readObject(in); |
| this.removeAllDataSize = (int) InternalDataSerializer.readUnsignedVL(in); |
| this.removeAllData = new RemoveAllEntryData[this.removeAllDataSize]; |
| if (this.removeAllDataSize > 0) { |
| final Version version = InternalDataSerializer.getVersionForDataStreamOrNull(in); |
| final ByteArrayDataInput bytesIn = new ByteArrayDataInput(); |
| for (int i = 0; i < this.removeAllDataSize; i++) { |
| this.removeAllData[i] = new RemoveAllEntryData(in, eventId, i, version, bytesIn); |
| } |
| |
| boolean hasTags = in.readBoolean(); |
| if (hasTags) { |
| EntryVersionsList versionTags = EntryVersionsList.create(in); |
| for (int i = 0; i < this.removeAllDataSize; i++) { |
| this.removeAllData[i].versionTag = versionTags.get(i); |
| } |
| } |
| } |
| |
| if ((flags & HAS_BRIDGE_CONTEXT) != 0) { |
| this.context = DataSerializer.readObject(in); |
| } |
| this.skipCallbacks = (flags & SKIP_CALLBACKS) != 0; |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| |
| super.toData(out); |
| DataSerializer.writeObject(this.eventId, out); |
| InternalDataSerializer.writeUnsignedVL(this.removeAllDataSize, out); |
| if (this.removeAllDataSize > 0) { |
| EntryVersionsList versionTags = new EntryVersionsList(removeAllDataSize); |
| |
| boolean hasTags = false; |
| for (int i = 0; i < this.removeAllDataSize; i++) { |
| if (!hasTags && removeAllData[i].versionTag != null) { |
| hasTags = true; |
| } |
| VersionTag<?> tag = removeAllData[i].versionTag; |
| versionTags.add(tag); |
| removeAllData[i].versionTag = null; |
| this.removeAllData[i].toData(out); |
| this.removeAllData[i].versionTag = tag; |
| } |
| |
| out.writeBoolean(hasTags); |
| if (hasTags) { |
| InternalDataSerializer.invokeToData(versionTags, out); |
| } |
| } |
| if (this.context != null) { |
| DataSerializer.writeObject(this.context, out); |
| } |
| } |
| |
| @Override |
| protected short computeCompressedShort(short s) { |
| s = super.computeCompressedShort(s); |
| if (this.context != null) |
| s |= HAS_BRIDGE_CONTEXT; |
| if (this.skipCallbacks) |
| s |= SKIP_CALLBACKS; |
| return s; |
| } |
| |
| public ClientProxyMembershipID getContext() { |
| return this.context; |
| } |
| |
| public RemoveAllEntryData[] getRemoveAllEntryData() { |
| return this.removeAllData; |
| } |
| |
| } |
| } |
