| /* |
| * 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.tx; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.distributed.DistributedSystemDisconnectedException; |
| 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.DistributionStats; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| 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.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.cache.BucketRegion.RawValue; |
| import org.apache.geode.internal.cache.CachedDeserializableFactory; |
| import org.apache.geode.internal.cache.DataLocationException; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.KeyInfo; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.RemoteOperationException; |
| import org.apache.geode.internal.cache.TXManagerImpl; |
| import org.apache.geode.internal.cache.TXStateProxy; |
| import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.offheap.OffHeapHelper; |
| import org.apache.geode.internal.serialization.DeserializationContext; |
| import org.apache.geode.internal.serialization.SerializationContext; |
| import org.apache.geode.internal.serialization.Version; |
| import org.apache.geode.internal.util.BlobHelper; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| /** |
| * This message is used as the request for a get operation done in a transaction that is hosted on a |
| * remote member. This messsage sends the get to the remote member. |
| * |
| * @since GemFire 6.5 |
| */ |
| public class RemoteGetMessage extends RemoteOperationMessageWithDirectReply { |
| private static final Logger logger = LogService.getLogger(); |
| |
| private Object key; |
| |
| /** The callback arg of the operation */ |
| private Object cbArg; |
| |
| private ClientProxyMembershipID context; |
| |
| |
| /** |
| * Empty constructor to satisfy {@link DataSerializer} requirements |
| */ |
| public RemoteGetMessage() {} |
| |
| private RemoteGetMessage(InternalDistributedMember recipient, String regionPath, |
| DirectReplyProcessor processor, final Object key, final Object aCallbackArgument, |
| ClientProxyMembershipID context) { |
| super(recipient, regionPath, processor); |
| this.key = key; |
| this.cbArg = aCallbackArgument; |
| this.context = context; |
| } |
| |
| @Override |
| protected boolean operateOnRegion(final ClusterDistributionManager dm, LocalRegion r, |
| long startTime) throws RemoteOperationException { |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, "RemoteGetMessage operateOnRegion: {}", r.getFullPath()); |
| } |
| |
| if (this.getTXUniqId() != TXManagerImpl.NOTX) { |
| assert r.getDataView() instanceof TXStateProxy; |
| } |
| |
| r.waitOnInitialization(); // bug #43371 - accessing a region before it's initialized |
| |
| RawValue valueBytes; |
| Object val = null; |
| try { |
| KeyInfo keyInfo = r.getKeyInfo(key, cbArg); |
| val = r.getDataView().getSerializedValue(r, keyInfo, false, this.context, null, |
| false /* for replicate regions */); |
| valueBytes = val instanceof RawValue ? (RawValue) val : new RawValue(val); |
| |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, |
| "GetMessage sending serialized value {} back via GetReplyMessage using processorId: {}", |
| valueBytes, getProcessorId()); |
| } |
| |
| GetReplyMessage.send(getSender(), getProcessorId(), valueBytes, getReplySender(dm)); |
| |
| // Unless an exception was thrown, this message handles sending the response |
| return false; |
| } catch (DistributedSystemDisconnectedException sde) { |
| sendReply(getSender(), this.processorId, dm, |
| new ReplyException(new RemoteOperationException( |
| "Operation got interrupted due to shutdown in progress on remote VM.", |
| sde)), |
| r, startTime); |
| return false; |
| } catch (DataLocationException e) { |
| sendReply(getSender(), getProcessorId(), dm, new ReplyException(e), r, startTime); |
| return false; |
| } finally { |
| OffHeapHelper.release(val); |
| } |
| |
| } |
| |
| @Override |
| protected void appendFields(StringBuffer buff) { |
| super.appendFields(buff); |
| buff.append("; key=").append(this.key).append("; callback arg=").append(this.cbArg); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return R_GET_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| this.key = DataSerializer.readObject(in); |
| this.cbArg = DataSerializer.readObject(in); |
| this.context = DataSerializer.readObject(in); |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| DataSerializer.writeObject(this.key, out); |
| DataSerializer.writeObject(this.cbArg, out); |
| DataSerializer.writeObject(this.context, out); |
| } |
| |
| public void setKey(Object key) { |
| this.key = key; |
| } |
| |
| /** |
| * Sends a ReplicateRegion {@link org.apache.geode.cache.Region#get(Object)} message |
| * |
| * @param recipient the member that the get message is sent to |
| * @param r the ReplicateRegion for which get was performed upon |
| * @param key the object to which the value should be feteched |
| * @param requestingClient the client requesting the value |
| * @return the processor used to fetch the returned value associated with the key |
| */ |
| public static RemoteGetResponse send(InternalDistributedMember recipient, LocalRegion r, |
| final Object key, final Object aCallbackArgument, ClientProxyMembershipID requestingClient) |
| throws RemoteOperationException { |
| Assert.assertTrue(recipient != null, "RemoteGetMessage NULL recipient"); |
| RemoteGetResponse p = new RemoteGetResponse(r.getSystem(), recipient); |
| RemoteGetMessage m = new RemoteGetMessage(recipient, r.getFullPath(), p, key, aCallbackArgument, |
| requestingClient); |
| Set<?> failures = r.getDistributionManager().putOutgoing(m); |
| if (failures != null && failures.size() > 0) { |
| throw new RemoteOperationException( |
| String.format("Failed sending < %s >", m)); |
| } |
| |
| return p; |
| } |
| |
| /** |
| * This message is used for the reply to a |
| * {@link org.apache.geode.cache.Region#get(Object)}operation This is the reply to a |
| * {@link RemoteGetMessage}. |
| * |
| * Since the {@link org.apache.geode.cache.Region#get(Object)}operation is used <bold>very </bold> |
| * frequently the performance of this class is critical. |
| * |
| * @since GemFire 6.5 |
| */ |
| public static class GetReplyMessage extends ReplyMessage { |
| /** |
| * The raw value in the cache which may be serialized to the output stream, if it is NOT already |
| * a byte array |
| */ |
| private transient RawValue rawVal; |
| |
| /** |
| * Indicates that the value already a byte array (aka user blob) and does not need |
| * de-serialization |
| */ |
| public boolean valueIsByteArray; |
| |
| /* |
| * Used on the fromData side to transfer the value bytes to the requesting thread |
| */ |
| public transient byte[] valueInBytes; |
| |
| public transient Version remoteVersion; |
| |
| /** |
| * Empty constructor to conform to DataSerializable interface |
| */ |
| public GetReplyMessage() {} |
| |
| private GetReplyMessage(int processorId, RawValue val) { |
| setProcessorId(processorId); |
| this.rawVal = val; |
| this.valueIsByteArray = val.isValueByteArray(); |
| } |
| |
| /** GetReplyMessages are always processed in-line */ |
| @Override |
| public boolean getInlineProcess() { |
| return true; |
| } |
| |
| /** |
| * Return the value from the get operation, serialize it bytes as late as possible to avoid |
| * making un-neccesary byte[] copies. De-serialize those same bytes as late as possible to avoid |
| * using precious threads (aka P2P readers). |
| * |
| * @param recipient the origin VM that performed the get |
| * @param processorId the processor on which the origin thread is waiting |
| * @param val the raw value that will eventually be serialized |
| * @param replySender distribution manager used to send the reply |
| */ |
| public static void send(InternalDistributedMember recipient, int processorId, RawValue val, |
| ReplySender replySender) throws RemoteOperationException { |
| Assert.assertTrue(recipient != null, "PRDistribuedGetReplyMessage NULL reply message"); |
| GetReplyMessage m = new GetReplyMessage(processorId, val); |
| m.setRecipient(recipient); |
| replySender.putOutgoing(m); |
| } |
| |
| /** |
| * Processes this message. This method is invoked by the receiver of the message. |
| * |
| * @param dm the distribution manager that is processing the message. |
| */ |
| @Override |
| public void process(final DistributionManager dm, ReplyProcessor21 processor) { |
| final boolean isDebugEnabled = logger.isTraceEnabled(LogMarker.DM_VERBOSE); |
| |
| final long startTime = getTimestamp(); |
| if (isDebugEnabled) { |
| logger.trace(LogMarker.DM_VERBOSE, |
| "GetReplyMessage process invoking reply processor with processorId:{}", |
| this.processorId); |
| } |
| |
| if (processor == null) { |
| if (isDebugEnabled) { |
| logger.trace(LogMarker.DM_VERBOSE, "GetReplyMessage processor not found"); |
| } |
| return; |
| } |
| processor.process(this); |
| |
| if (isDebugEnabled) { |
| logger.trace(LogMarker.DM_VERBOSE, "{} Processed {}", processor, this); |
| } |
| dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return R_GET_REPLY_MESSAGE; |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| out.writeByte(this.valueIsByteArray ? 1 : 0); |
| this.rawVal.writeAsByteArray(out); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| this.valueIsByteArray = (in.readByte() == 1); |
| this.valueInBytes = DataSerializer.readByteArray(in); |
| if (!this.valueIsByteArray) { |
| this.remoteVersion = InternalDataSerializer.getVersionForDataStreamOrNull(in); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| StringBuffer sb = new StringBuffer(); |
| sb.append("GetReplyMessage ").append("processorid=").append(this.processorId) |
| .append(" reply to sender ").append(this.getSender()) |
| .append(" returning serialized value=").append(this.rawVal); |
| return sb.toString(); |
| } |
| } |
| |
| /** |
| * A processor to capture the value returned by |
| * {@link org.apache.geode.internal.cache.tx.RemoteGetMessage.GetReplyMessage} |
| * |
| * @since GemFire 5.0 |
| */ |
| public static class RemoteGetResponse extends RemoteOperationResponse { |
| |
| private volatile GetReplyMessage getReply; |
| private volatile boolean returnValueReceived; |
| private volatile long start; |
| |
| public RemoteGetResponse(InternalDistributedSystem ds, InternalDistributedMember recipient) { |
| super(ds, recipient, false); |
| } |
| |
| @Override |
| public void process(DistributionMessage msg) { |
| if (DistributionStats.enableClockStats) { |
| this.start = DistributionStats.getStatTime(); |
| } |
| if (msg instanceof GetReplyMessage) { |
| GetReplyMessage reply = (GetReplyMessage) msg; |
| // De-serialization needs to occur in the requesting thread, not a P2P thread |
| // (or some other limited resource) |
| if (reply.valueInBytes != null) { |
| this.getReply = reply; |
| } |
| this.returnValueReceived = true; |
| } |
| super.process(msg); |
| } |
| |
| /** |
| * De-seralize the value, if the value isn't already a byte array, this method should be called |
| * in the context of the requesting thread for the best scalability |
| * |
| * @see EntryEventImpl#deserialize(byte[]) |
| * @return the value object |
| */ |
| public Object getValue(boolean preferCD) throws RemoteOperationException { |
| final GetReplyMessage reply = this.getReply; |
| try { |
| if (reply != null) { |
| if (reply.valueIsByteArray) { |
| return reply.valueInBytes; |
| } else if (preferCD) { |
| return CachedDeserializableFactory.create(reply.valueInBytes, |
| getDistributionManager().getCache()); |
| } else { |
| return BlobHelper.deserializeBlob(reply.valueInBytes, reply.remoteVersion, null); |
| } |
| } |
| return null; |
| } catch (IOException e) { |
| throw new RemoteOperationException( |
| "Unable to deserialize value (IOException)", |
| e); |
| } catch (ClassNotFoundException e) { |
| throw new RemoteOperationException( |
| "Unable to deserialize value (ClassNotFoundException)", |
| e); |
| } |
| } |
| |
| |
| /** |
| * @return Object associated with the key that was sent in the get message |
| */ |
| public Object waitForResponse(boolean preferCD) throws RemoteOperationException { |
| waitForRemoteResponse(); |
| if (DistributionStats.enableClockStats) { |
| getDistributionManager().getStats().incReplyHandOffTime(this.start); |
| } |
| if (!this.returnValueReceived) { |
| throw new RemoteOperationException( |
| "no return value received"); |
| } |
| return getValue(preferCD); |
| } |
| } |
| |
| } |