| /* |
| * 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.Externalizable; |
| import java.io.IOException; |
| import java.io.ObjectInput; |
| import java.io.ObjectOutput; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Set; |
| |
| import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; |
| 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.DistributedMember; |
| 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.DataSerializableFixedID; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.Version; |
| 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.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.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.logging.log4j.LogMarker; |
| 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.putall operation. |
| * |
| * @since GemFire 5.0 |
| */ |
| public class DistributedPutAllOperation extends AbstractUpdateOperation { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| protected final PutAllEntryData[] putAllData; |
| |
| public int putAllDataSize; |
| |
| 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; |
| |
| // flags for CachedDeserializable; additional flags can be combined |
| // with these if required |
| static final byte IS_CACHED_DESER = 0x1; |
| static final byte IS_OBJECT = 0x2; |
| |
| // private boolean containsCreate = false; |
| |
| public DistributedPutAllOperation(CacheEvent event, int size, boolean isBridgeOp) { |
| super(event, ((EntryEventImpl) event).getEventTime(0L)); |
| this.putAllData = new PutAllEntryData[size]; |
| this.putAllDataSize = 0; |
| this.isBridgeOp = isBridgeOp; |
| } |
| |
| /** |
| * return if the operation is bridge operation |
| */ |
| public boolean isBridgeOperation() { |
| return this.isBridgeOp; |
| } |
| |
| public PutAllEntryData[] getPutAllEntryData() { |
| return putAllData; |
| } |
| |
| public void setPutAllEntryData(PutAllEntryData[] putAllEntryData) { |
| for (int i = 0; i < putAllEntryData.length; i++) { |
| putAllData[i] = putAllEntryData[i]; |
| } |
| this.putAllDataSize = putAllEntryData.length; |
| } |
| |
| /** |
| * Add an entry that this putall operation should distribute. |
| */ |
| public void addEntry(PutAllEntryData putAllEntry) { |
| this.putAllData[this.putAllDataSize] = putAllEntry; |
| this.putAllDataSize += 1; |
| // cachedEvents.add(ev); |
| } |
| |
| /** |
| * Add an entry that this putall operation should distribute. |
| */ |
| public void addEntry(EntryEventImpl ev) { |
| this.putAllData[this.putAllDataSize] = new PutAllEntryData(ev); |
| this.putAllDataSize += 1; |
| // cachedEvents.add(ev); |
| } |
| |
| /** |
| * Add an entry that this putall 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.putAllData[this.putAllDataSize] = new PutAllEntryData(ev); |
| this.putAllData[this.putAllDataSize].setCallbacksInvoked(newCallbackInvoked); |
| this.putAllDataSize += 1; |
| // cachedEvents.add(ev); |
| } |
| |
| /** |
| * 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.putAllData[this.putAllDataSize] = new PutAllEntryData(ev); |
| this.putAllData[this.putAllDataSize].setBucketId(bucketId); |
| this.putAllDataSize += 1; |
| // cachedEvents.add(ev); |
| } |
| |
| /** |
| * 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 < putAllDataSize; i++) { |
| putAllData[i].setUsedFakeEventId(status); |
| } |
| } |
| |
| /** |
| * In the originating cache, this returns an iterator on the list of events caused by the putAll |
| * 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 DistributedPutAllOperation.this.putAllDataSize > 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.putAllDataSize; i++) { |
| PutAllEntryData entry = this.putAllData[i]; |
| if (entry != null && entry.event != null) { |
| entry.event.release(); |
| } |
| } |
| } |
| |
| @Unretained |
| public EntryEventImpl getEventForPosition(int position) { |
| PutAllEntryData entry = this.putAllData[position]; |
| if (entry == null) { |
| return null; |
| } |
| if (entry.event != null) { |
| return entry.event; |
| } |
| LocalRegion region = (LocalRegion) this.event.getRegion(); |
| @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()); |
| 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; |
| Object entryValue = entry.getValue(region.getCache()); |
| if (entryValue == null |
| && ev.getRegion().getAttributes().getDataPolicy() == DataPolicy.NORMAL) { |
| ev.setLocalInvalid(true); |
| } |
| ev.setNewValue(entryValue); |
| 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 putall'd. |
| */ |
| public static class PutAllEntryData { |
| |
| final Object key; |
| |
| private Object value; |
| |
| 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; |
| |
| /** |
| * Constructor to use when preparing to send putall data out |
| */ |
| public PutAllEntryData(EntryEventImpl event) { |
| |
| this.key = event.getKey(); |
| this.value = event.getRawNewValueAsHeapObject(); |
| Object oldValue = event.getRawOldValueAsHeapObject(); |
| |
| 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()); |
| } |
| |
| /** |
| * Constructor to use when receiving a putall from someone else |
| */ |
| public PutAllEntryData(DataInput in, EventID baseEventID, int idx, Version version, |
| ByteArrayDataInput bytesIn) throws IOException, ClassNotFoundException { |
| this.key = DataSerializer.readObject(in); |
| byte flgs = in.readByte(); |
| if ((flgs & IS_OBJECT) != 0) { |
| this.value = DataSerializer.readObject(in); |
| } else { |
| byte[] bb = DataSerializer.readByteArray(in); |
| if ((flgs & IS_CACHED_DESER) != 0) { |
| this.value = new FutureCachedDeserializable(bb); |
| } else { |
| this.value = bb; |
| } |
| } |
| 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(this.value).append(",") |
| .append(getOldValue()); |
| if (this.bucketId > 0) { |
| sb.append(", b").append(this.bucketId); |
| } |
| if (versionTag != null) { |
| sb.append(versionTag); |
| // 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 PutAllMessage#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; |
| final Object v = this.value; |
| DataSerializer.writeObject(key, out); |
| |
| if (v instanceof byte[] || v == null) { |
| out.writeByte(0); |
| DataSerializer.writeByteArray((byte[]) v, out); |
| } else if (v instanceof CachedDeserializable) { |
| CachedDeserializable cd = (CachedDeserializable) v; |
| out.writeByte(IS_CACHED_DESER); |
| DataSerializer.writeByteArray(cd.getSerializedValue(), out); |
| } else { |
| out.writeByte(IS_CACHED_DESER); |
| DataSerializer.writeObjectAsByteArray(v, 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 value |
| */ |
| public Object getValue(InternalCache cache) { |
| Object result = this.value; |
| if (result instanceof FutureCachedDeserializable) { |
| FutureCachedDeserializable future = (FutureCachedDeserializable) result; |
| result = future.create(cache); |
| this.value = result; |
| } |
| return result; |
| } |
| |
| /** |
| * 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 isCallbacksInvoked() { |
| return this.callbacksInvoked; |
| } |
| |
| public void setCallbacksInvoked(boolean callbacksInvoked) { |
| this.callbacksInvoked = callbacksInvoked; |
| } |
| } |
| |
| public static class EntryVersionsList extends ArrayList<VersionTag> |
| implements DataSerializableFixedID, Externalizable { |
| |
| public EntryVersionsList() { |
| // Do nothing |
| } |
| |
| public EntryVersionsList(int size) { |
| super(size); |
| } |
| |
| public static EntryVersionsList create(DataInput in) |
| throws IOException, ClassNotFoundException { |
| EntryVersionsList newList = new EntryVersionsList(); |
| InternalDataSerializer.invokeFromData(newList, in); |
| return newList; |
| } |
| |
| private boolean extractVersion(PutAllEntryData entry) { |
| |
| VersionTag versionTag = entry.versionTag; |
| // version tag can be null if only keys are sent in InitialImage. |
| if (versionTag != null) { |
| add(versionTag); |
| // Add entry without version tag in entries array. |
| entry.versionTag = null; |
| return true; |
| } |
| |
| return false; |
| } |
| |
| private VersionTag<VersionSource> getVersionTag(int index) { |
| VersionTag tag = null; |
| if (this.size() > 0) { |
| tag = get(index); |
| } |
| return tag; |
| } |
| |
| /** |
| * replace null membership IDs in version tags with the given member ID. VersionTags received |
| * from a server may have null IDs because they were operations performed by that server. We |
| * transmit them as nulls to cut costs, but have to do the swap on the receiving end (in the |
| * client) |
| * |
| */ |
| public void replaceNullIDs(DistributedMember sender) { |
| for (VersionTag versionTag : this) { |
| if (versionTag != null) { |
| versionTag.replaceNullIDs((InternalDistributedMember) sender); |
| } |
| } |
| } |
| |
| @Override |
| public int getDSFID() { |
| return DataSerializableFixedID.PUTALL_VERSIONS_LIST; |
| } |
| |
| static final byte FLAG_NULL_TAG = 0; |
| static final byte FLAG_FULL_TAG = 1; |
| static final byte FLAG_TAG_WITH_NEW_ID = 2; |
| static final byte FLAG_TAG_WITH_NUMBER_ID = 3; |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| int flags = 0; |
| boolean hasTags = false; |
| |
| if (this.size() > 0) { |
| flags |= 0x04; |
| hasTags = true; |
| for (VersionTag tag : this) { |
| if (tag != null) { |
| if (tag instanceof DiskVersionTag) { |
| flags |= 0x20; |
| } |
| break; |
| } |
| } |
| } |
| |
| if (logger.isTraceEnabled(LogMarker.INITIAL_IMAGE_VERSIONED_VERBOSE)) { |
| logger.trace(LogMarker.INITIAL_IMAGE_VERSIONED_VERBOSE, "serializing {} with flags 0x{}", |
| this, Integer.toHexString(flags)); |
| } |
| |
| out.writeByte(flags); |
| |
| if (hasTags) { |
| InternalDataSerializer.writeUnsignedVL(this.size(), out); |
| Object2IntOpenHashMap ids = new Object2IntOpenHashMap(this.size()); |
| int idCount = 0; |
| for (VersionTag tag : this) { |
| if (tag == null) { |
| out.writeByte(FLAG_NULL_TAG); |
| } else { |
| VersionSource id = tag.getMemberID(); |
| if (id == null) { |
| out.writeByte(FLAG_FULL_TAG); |
| InternalDataSerializer.invokeToData(tag, out); |
| } else { |
| int idNumber = ids.getInt(id); |
| if (idNumber == 0) { |
| out.writeByte(FLAG_TAG_WITH_NEW_ID); |
| idNumber = ++idCount; |
| ids.put(id, idNumber); |
| InternalDataSerializer.invokeToData(tag, out); |
| } else { |
| out.writeByte(FLAG_TAG_WITH_NUMBER_ID); |
| tag.toData(out, false); |
| tag.setMemberID(id); |
| InternalDataSerializer.writeUnsignedVL(idNumber - 1, out); |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| int flags = in.readByte(); |
| boolean hasTags = (flags & 0x04) == 0x04; |
| boolean persistent = (flags & 0x20) == 0x20; |
| |
| if (logger.isTraceEnabled(LogMarker.INITIAL_IMAGE_VERSIONED_VERBOSE)) { |
| logger.trace(LogMarker.INITIAL_IMAGE_VERSIONED_VERBOSE, |
| "deserializing a InitialImageVersionedObjectList with flags 0x{}", |
| Integer.toHexString(flags)); |
| } |
| |
| if (hasTags) { |
| int size = (int) InternalDataSerializer.readUnsignedVL(in); |
| if (logger.isTraceEnabled(LogMarker.INITIAL_IMAGE_VERSIONED_VERBOSE)) { |
| logger.trace(LogMarker.INITIAL_IMAGE_VERSIONED_VERBOSE, "reading {} version tags", size); |
| } |
| List<VersionSource> ids = new ArrayList<VersionSource>(size); |
| for (int i = 0; i < size; i++) { |
| byte entryType = in.readByte(); |
| switch (entryType) { |
| case FLAG_NULL_TAG: |
| add(null); |
| break; |
| case FLAG_FULL_TAG: |
| add(VersionTag.create(persistent, in)); |
| break; |
| case FLAG_TAG_WITH_NEW_ID: |
| VersionTag tag = VersionTag.create(persistent, in); |
| ids.add(tag.getMemberID()); |
| add(tag); |
| break; |
| case FLAG_TAG_WITH_NUMBER_ID: |
| tag = VersionTag.create(persistent, in); |
| int idNumber = (int) InternalDataSerializer.readUnsignedVL(in); |
| tag.setMemberID(ids.get(idNumber)); |
| add(tag); |
| break; |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void writeExternal(ObjectOutput out) throws IOException { |
| toData(out); |
| } |
| |
| @Override |
| public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { |
| fromData(in); |
| } |
| |
| @Override |
| public Version[] getSerializationVersions() { |
| return null; |
| } |
| } |
| |
| @Override |
| protected FilterRoutingInfo getRecipientFilterRouting(Set cacheOpRecipients) { |
| // for putAll, 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.putAllData.length; i++) { |
| @Unretained |
| EntryEventImpl ev = getEventForPosition(i); |
| if (ev != null) { |
| FilterRoutingInfo eventRouting = advisor.adviseFilterRouting(ev, cacheOpRecipients); |
| if (eventRouting != null) { |
| consolidated.addFilterInfo(eventRouting); |
| } |
| putAllData[i].filterRouting = eventRouting; |
| } |
| } |
| // we need to create routing information for each PUT event |
| return consolidated; |
| } |
| |
| |
| @Override |
| protected FilterInfo getLocalFilterRouting(FilterRoutingInfo frInfo) { |
| // long start = NanoTimer.getTime(); |
| FilterProfile fp = getRegion().getFilterProfile(); |
| if (fp == null) { |
| return null; |
| } |
| |
| // this will set the local FilterInfo in the events |
| if (this.putAllData != null && this.putAllData.length > 0) { |
| fp.getLocalFilterRoutingForPutAllOp(this, this.putAllData); |
| } |
| |
| // long finish = NanoTimer.getTime(); |
| // InternalDistributedSystem.getDMStats().incjChannelUpTime(finish-start); |
| return null; |
| } |
| |
| @Override |
| protected CacheOperationMessage createMessage() { |
| EntryEventImpl event = getBaseEvent(); |
| PutAllMessage msg = new PutAllMessage(); |
| msg.eventId = event.getEventId(); |
| msg.context = event.getContext(); |
| return msg; |
| } |
| |
| /** |
| * Create PutAllPRMessage for notify only (to adjunct nodes) |
| * |
| * @param bucketId create message to send to this bucket |
| */ |
| public PutAllPRMessage createPRMessagesNotifyOnly(int bucketId) { |
| final EntryEventImpl event = getBaseEvent(); |
| PutAllPRMessage prMsg = new PutAllPRMessage(bucketId, putAllDataSize, 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 < putAllDataSize; i++) { |
| prMsg.addEntry(putAllData[i]); |
| } |
| |
| return prMsg; |
| } |
| |
| /** |
| * Create PutAllPRMessages for primary buckets out of dpao |
| * |
| * @return a HashMap contain PutAllPRMessages, key is bucket id |
| */ |
| public HashMap createPRMessages() { |
| // getFilterRecipients(Collections.EMPTY_SET); // establish filter recipient routing information |
| HashMap prMsgMap = new HashMap(); |
| final EntryEventImpl event = getBaseEvent(); |
| |
| for (int i = 0; i < putAllDataSize; i++) { |
| Integer bucketId = putAllData[i].bucketId; |
| PutAllPRMessage prMsg = (PutAllPRMessage) prMsgMap.get(bucketId); |
| if (prMsg == null) { |
| prMsg = new PutAllPRMessage(bucketId.intValue(), putAllDataSize, 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. |
| putAllData[i].setFakeEventID(); |
| // we only save the reference in prMsg. No duplicate copy |
| prMsg.addEntry(putAllData[i]); |
| prMsgMap.put(bucketId, prMsg); |
| } |
| return prMsgMap; |
| } |
| |
| @Override |
| protected void initMessage(CacheOperationMessage msg, DirectReplyProcessor proc) { |
| super.initMessage(msg, proc); |
| PutAllMessage m = (PutAllMessage) 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 = RemotePutAllMessage.distribute((EntryEventImpl) this.event, |
| this.putAllData, this.putAllDataSize); |
| if (success) { |
| m.callbackArg = this.event.getCallbackArgument(); |
| m.putAllData = new PutAllEntryData[0]; |
| m.putAllDataSize = 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 |
| PutAllEntryData[] versionless = selectVersionlessEntries(); |
| if (logger.isTraceEnabled()) { |
| logger.trace("Found these versionless entries: {}", Arrays.toString(versionless)); |
| } |
| if (versionless.length > 0) { |
| boolean success = RemotePutAllMessage.distribute((EntryEventImpl) this.event, versionless, |
| versionless.length); |
| if (success) { |
| versionless = null; |
| PutAllEntryData[] versioned = selectVersionedEntries(); |
| if (logger.isTraceEnabled()) { |
| logger.trace("Found these remaining versioned entries: {}", |
| Arrays.toString(versioned)); |
| } |
| m.callbackArg = this.event.getCallbackArgument(); |
| m.putAllData = versioned; |
| m.putAllDataSize = 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.putAllData = this.putAllData; |
| m.putAllDataSize = this.putAllDataSize; |
| 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 PutAllEntryData[] selectVersionlessEntries() { |
| int resultSize = this.putAllData.length; |
| for (int i = 0; i < this.putAllData.length; i++) { |
| PutAllEntryData p = this.putAllData[i]; |
| if (p == null || p.isInhibitDistribution()) { |
| resultSize--; |
| } else if (p.versionTag != null && p.versionTag.hasValidVersion()) { |
| resultSize--; |
| } |
| } |
| PutAllEntryData[] result = new PutAllEntryData[resultSize]; |
| int ri = 0; |
| for (int i = 0; i < this.putAllData.length; i++) { |
| PutAllEntryData p = this.putAllData[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 PutAllEntryData[] selectVersionedEntries() { |
| int resultSize = 0; |
| for (int i = 0; i < this.putAllData.length; i++) { |
| PutAllEntryData p = this.putAllData[i]; |
| if (p == null || p.isInhibitDistribution()) { |
| continue; // skip blanks |
| } else if (p.versionTag != null && p.versionTag.hasValidVersion()) { |
| resultSize++; |
| } |
| } |
| PutAllEntryData[] result = new PutAllEntryData[resultSize]; |
| int ri = 0; |
| for (int i = 0; i < this.putAllData.length; i++) { |
| PutAllEntryData p = this.putAllData[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 (PutAllEntryData entry : this.putAllData) { |
| if (entry.versionTag != null) { |
| list.addKeyAndVersion(entry.key, entry.versionTag); |
| } |
| } |
| } |
| |
| |
| public static class PutAllMessage extends AbstractUpdateMessage { |
| |
| protected PutAllEntryData[] putAllData; |
| |
| protected int putAllDataSize; |
| |
| 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.putAllData.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.PUTALL_UPDATE /* op */, |
| 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.putAllDataSize); |
| if (putAllDataSize <= 20) { |
| // 20 is a size for test |
| sb.append("; entry values=").append(Arrays.toString(this.putAllData)); |
| } |
| } |
| |
| /** |
| * Does the "put" of one entry for a "putall" operation. Note it calls back to |
| * AbstractUpdateOperation.UpdateMessage#basicOperationOnRegion |
| * |
| * @param entry the entry being put |
| * @param rgn the region the entry is put in |
| */ |
| public void doEntryPut(PutAllEntryData entry, DistributedRegion rgn) { |
| @Released |
| EntryEventImpl ev = PutAllMessage.createEntryEvent(entry, getSender(), this.context, rgn, |
| this.possibleDuplicate, this.needsRouting, this.callbackArg, true, skipCallbacks); |
| // we don't need to set old value here, because the msg is from remote. local old value will |
| // get from next step |
| try { |
| super.basicOperateOnRegion(ev, rgn); |
| } finally { |
| if (ev.hasValidVersionTag() && !ev.getVersionTag().isRecorded()) { |
| if (rgn.getVersionVector() != null) { |
| rgn.getVersionVector().recordVersion(getSender(), ev.getVersionTag()); |
| } |
| } |
| ev.release(); |
| } |
| } |
| |
| /** |
| * create an event for a PutAllEntryData element |
| * |
| * @return the event to be used in applying the element |
| */ |
| @Retained |
| public static EntryEventImpl createEntryEvent(PutAllEntryData 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; |
| } |
| Object entryValue = entry.getValue(rgn.getCache()); |
| if (entryValue == null && rgn.getDataPolicy() == DataPolicy.NORMAL) { |
| ev.setLocalInvalid(true); |
| } |
| ev.setNewValue(entryValue); |
| 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 < putAllDataSize; ++i) { |
| if (putAllData[i].versionTag != null) { |
| checkVersionTag(rgn, putAllData[i].versionTag); |
| } |
| } |
| |
| rgn.syncBulkOp(new Runnable() { |
| @Override |
| public void run() { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| for (int i = 0; i < putAllDataSize; ++i) { |
| if (isDebugEnabled) { |
| logger.debug("putAll processing {} with {} sender={}", putAllData[i], |
| putAllData[i].versionTag, sender); |
| } |
| putAllData[i].setSender(sender); |
| doEntryPut(putAllData[i], rgn); |
| } |
| } |
| }, ev.getEventId()); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return PUT_ALL_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| |
| super.fromData(in); |
| this.eventId = (EventID) DataSerializer.readObject(in); |
| this.putAllDataSize = (int) InternalDataSerializer.readUnsignedVL(in); |
| this.putAllData = new PutAllEntryData[this.putAllDataSize]; |
| if (this.putAllDataSize > 0) { |
| final Version version = InternalDataSerializer.getVersionForDataStreamOrNull(in); |
| final ByteArrayDataInput bytesIn = new ByteArrayDataInput(); |
| for (int i = 0; i < this.putAllDataSize; i++) { |
| this.putAllData[i] = new PutAllEntryData(in, eventId, i, version, bytesIn); |
| } |
| |
| boolean hasTags = in.readBoolean(); |
| if (hasTags) { |
| EntryVersionsList versionTags = EntryVersionsList.create(in); |
| for (int i = 0; i < this.putAllDataSize; i++) { |
| this.putAllData[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.putAllDataSize, out); |
| if (this.putAllDataSize > 0) { |
| EntryVersionsList versionTags = new EntryVersionsList(putAllDataSize); |
| |
| boolean hasTags = false; |
| for (int i = 0; i < this.putAllDataSize; i++) { |
| if (!hasTags && putAllData[i].versionTag != null) { |
| hasTags = true; |
| } |
| VersionTag<?> tag = putAllData[i].versionTag; |
| versionTags.add(tag); |
| putAllData[i].versionTag = null; |
| this.putAllData[i].toData(out); |
| this.putAllData[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 PutAllEntryData[] getPutAllEntryData() { |
| return this.putAllData; |
| } |
| |
| } |
| } |
