<commit_before>/*=========================================================================
 * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
 * This product is protected by U.S. and international copyright
 * and intellectual property laws. Pivotal products are covered by
 * more patents listed at http://www.pivotal.io/patents.
 *=========================================================================
 */

package com.gemstone.gemfire.internal.cache.partitioned;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collections;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import com.gemstone.gemfire.DataSerializer;
import com.gemstone.gemfire.InternalGemFireError;
import com.gemstone.gemfire.cache.EntryNotFoundException;
import com.gemstone.gemfire.cache.TransactionDataNotColocatedException;
import com.gemstone.gemfire.distributed.DistributedSystemDisconnectedException;
import com.gemstone.gemfire.distributed.internal.DM;
import com.gemstone.gemfire.distributed.internal.DirectReplyProcessor;
import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.distributed.internal.DistributionMessage;
import com.gemstone.gemfire.distributed.internal.DistributionStats;
import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
import com.gemstone.gemfire.distributed.internal.ReplyException;
import com.gemstone.gemfire.distributed.internal.ReplyMessage;
import com.gemstone.gemfire.distributed.internal.ReplyProcessor21;
import com.gemstone.gemfire.distributed.internal.ReplySender;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
import com.gemstone.gemfire.internal.Assert;
import com.gemstone.gemfire.internal.InternalDataSerializer;
import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.cache.BucketRegion;
import com.gemstone.gemfire.internal.cache.BucketRegion.RawValue;
import com.gemstone.gemfire.internal.cache.CachedDeserializableFactory;
import com.gemstone.gemfire.internal.cache.DataLocationException;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.ForceReattemptException;
import com.gemstone.gemfire.internal.cache.KeyInfo;
import com.gemstone.gemfire.internal.cache.KeyWithRegionContext;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegionDataStore;
import com.gemstone.gemfire.internal.cache.PrimaryBucketException;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
import com.gemstone.gemfire.internal.cache.TXStateProxy;
import com.gemstone.gemfire.internal.cache.Token;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
import com.gemstone.gemfire.internal.offheap.OffHeapHelper;
import com.gemstone.gemfire.internal.util.BlobHelper;

/**
 * This message is used as the request for a
 * {@link com.gemstone.gemfire.cache.Region#get(Object)}operation. The reply is
 * sent in a {@link com.gemstone.gemfire.internal.cache.partitioned.GetMessage}. 
 * 
 * Since the {@link com.gemstone.gemfire.cache.Region#get(Object)}operation is
 * used <bold>very </bold> frequently the performance of this class is critical.
 * 
 * @author <NAME>
 * @since 5.0
 */
public final class GetMessage extends PartitionMessageWithDirectReply
  {
  private static final Logger logger = LogService.getLogger();
  
  private Object key;

  /** The callback arg of the operation */
  private Object cbArg;
  
  /** set to true if the message was requeued in the pr executor pool */
  private transient boolean forceUseOfPRExecutor;

  private ClientProxyMembershipID context;
  
  private boolean returnTombstones;

  private boolean allowReadFromHDFS;
  // reuse some flags
  protected static final int HAS_LOADER = NOTIFICATION_ONLY;
  protected static final int CAN_START_TX = IF_NEW;
  protected static final int READ_FROM_HDFS = IF_OLD;

  /**
   * Empty constructor to satisfy {@link DataSerializer} requirements
   */
  public GetMessage() {
  }
  
  private GetMessage(InternalDistributedMember recipient, int regionId,
      DirectReplyProcessor processor,
      final Object key, final Object aCallbackArgument, ClientProxyMembershipID context,
      boolean returnTombstones, boolean allowReadFromHDFS) {
    super(recipient, regionId, processor);
    this.key = key;
    this.cbArg = aCallbackArgument;
    this.context = context;
    this.returnTombstones = returnTombstones;
	this.allowReadFromHDFS = allowReadFromHDFS;
  }

  private static final boolean ORDER_PR_GETS = Boolean.getBoolean("gemfire.order-pr-gets");

  @Override
  final public int getProcessorType()
  {
    if (!forceUseOfPRExecutor && !ORDER_PR_GETS && !isDirectAck()) {
      try {
        PartitionedRegion pr = PartitionedRegion.getPRFromId(this.regionId);
        // If the region is persistent the get may need to fault a value
        // in which has to sync the region entry. Note it may need to
        // do this even if it is not overflow (after recovery values are faulted in async).
        // If the region has an LRU then in lruUpdateCallback it will
        // call getLRUEntry which has to sync a region entry.
        // Syncing a region entry can lead to dead-lock (see bug 52078).
        // So if either of these conditions hold this message can not be
        // processed in the p2p msg reader.
        if (pr.getAttributes().getDataPolicy().withPersistence()
            || !pr.getAttributes().getEvictionAttributes().getAlgorithm().isNone()) {
          return DistributionManager.PARTITIONED_REGION_EXECUTOR;
        }
      } catch (PRLocallyDestroyedException ignore) {
      }
    }
    if (forceUseOfPRExecutor) {
      return DistributionManager.PARTITIONED_REGION_EXECUTOR;
    }
    else if (ORDER_PR_GETS) {
      return DistributionManager.PARTITIONED_REGION_EXECUTOR;
    } else {
      // Make this guy serial so that it will be processed in the p2p msg reader
      // which gives it better performance.
      return DistributionManager.SERIAL_EXECUTOR;
    }
  }

  @Override
  public boolean isSevereAlertCompatible() {
    // allow forced-disconnect processing for all cache op messages
    return true;
  }
  
  @Override
  protected final boolean operateOnPartitionedRegion(
      final DistributionManager dm, PartitionedRegion r, long startTime)
      throws ForceReattemptException
  {
    if (logger.isTraceEnabled(LogMarker.DM)) {<commit_msg>GEODE-216: Handle any exception that might be thrown<commit_after>/*=========================================================================
 * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
 * This product is protected by U.S. and international copyright
 * and intellectual property laws. Pivotal products are covered by
 * more patents listed at http://www.pivotal.io/patents.
 *=========================================================================
 */

package com.gemstone.gemfire.internal.cache.partitioned;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collections;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import com.gemstone.gemfire.DataSerializer;
import com.gemstone.gemfire.InternalGemFireError;
import com.gemstone.gemfire.cache.EntryNotFoundException;
import com.gemstone.gemfire.cache.TransactionDataNotColocatedException;
import com.gemstone.gemfire.distributed.DistributedSystemDisconnectedException;
import com.gemstone.gemfire.distributed.internal.DM;
import com.gemstone.gemfire.distributed.internal.DirectReplyProcessor;
import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.distributed.internal.DistributionMessage;
import com.gemstone.gemfire.distributed.internal.DistributionStats;
import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
import com.gemstone.gemfire.distributed.internal.ReplyException;
import com.gemstone.gemfire.distributed.internal.ReplyMessage;
import com.gemstone.gemfire.distributed.internal.ReplyProcessor21;
import com.gemstone.gemfire.distributed.internal.ReplySender;
import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
import com.gemstone.gemfire.internal.Assert;
import com.gemstone.gemfire.internal.InternalDataSerializer;
import com.gemstone.gemfire.internal.Version;
import com.gemstone.gemfire.internal.cache.BucketRegion;
import com.gemstone.gemfire.internal.cache.BucketRegion.RawValue;
import com.gemstone.gemfire.internal.cache.CachedDeserializableFactory;
import com.gemstone.gemfire.internal.cache.DataLocationException;
import com.gemstone.gemfire.internal.cache.EntryEventImpl;
import com.gemstone.gemfire.internal.cache.ForceReattemptException;
import com.gemstone.gemfire.internal.cache.KeyInfo;
import com.gemstone.gemfire.internal.cache.KeyWithRegionContext;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;
import com.gemstone.gemfire.internal.cache.PartitionedRegionDataStore;
import com.gemstone.gemfire.internal.cache.PrimaryBucketException;
import com.gemstone.gemfire.internal.cache.TXManagerImpl;
import com.gemstone.gemfire.internal.cache.TXStateProxy;
import com.gemstone.gemfire.internal.cache.Token;
import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
import com.gemstone.gemfire.internal.cache.versions.VersionTag;
import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.internal.logging.log4j.LogMarker;
import com.gemstone.gemfire.internal.offheap.OffHeapHelper;
import com.gemstone.gemfire.internal.util.BlobHelper;

/**
 * This message is used as the request for a
 * {@link com.gemstone.gemfire.cache.Region#get(Object)}operation. The reply is
 * sent in a {@link com.gemstone.gemfire.internal.cache.partitioned.GetMessage}. 
 * 
 * Since the {@link com.gemstone.gemfire.cache.Region#get(Object)}operation is
 * used <bold>very </bold> frequently the performance of this class is critical.
 * 
 * @author <NAME>
 * @since 5.0
 */
public final class GetMessage extends PartitionMessageWithDirectReply
  {
  private static final Logger logger = LogService.getLogger();
  
  private Object key;

  /** The callback arg of the operation */
  private Object cbArg;
  
  /** set to true if the message was requeued in the pr executor pool */
  private transient boolean forceUseOfPRExecutor;

  private ClientProxyMembershipID context;
  
  private boolean returnTombstones;

  private boolean allowReadFromHDFS;
  // reuse some flags
  protected static final int HAS_LOADER = NOTIFICATION_ONLY;
  protected static final int CAN_START_TX = IF_NEW;
  protected static final int READ_FROM_HDFS = IF_OLD;

  /**
   * Empty constructor to satisfy {@link DataSerializer} requirements
   */
  public GetMessage() {
  }
  
  private GetMessage(InternalDistributedMember recipient, int regionId,
      DirectReplyProcessor processor,
      final Object key, final Object aCallbackArgument, ClientProxyMembershipID context,
      boolean returnTombstones, boolean allowReadFromHDFS) {
    super(recipient, regionId, processor);
    this.key = key;
    this.cbArg = aCallbackArgument;
    this.context = context;
    this.returnTombstones = returnTombstones;
	this.allowReadFromHDFS = allowReadFromHDFS;
  }

  private static final boolean ORDER_PR_GETS = Boolean.getBoolean("gemfire.order-pr-gets");

  @Override
  final public int getProcessorType()
  {
    if (!forceUseOfPRExecutor && !ORDER_PR_GETS && !isDirectAck()) {
      try {
        PartitionedRegion pr = PartitionedRegion.getPRFromId(this.regionId);
        // If the region is persistent the get may need to fault a value
        // in which has to sync the region entry. Note it may need to
        // do this even if it is not overflow (after recovery values are faulted in async).
        // If the region has an LRU then in lruUpdateCallback it will
        // call getLRUEntry which has to sync a region entry.
        // Syncing a region entry can lead to dead-lock (see bug 52078).
        // So if either of these conditions hold this message can not be
        // processed in the p2p msg reader.
        if (pr.getAttributes().getDataPolicy().withPersistence()
            || !pr.getAttributes().getEvictionAttributes().getAlgorithm().isNone()) {
          return DistributionManager.PARTITIONED_REGION_EXECUTOR;
        }
      } catch (PRLocallyDestroyedException ignore) {
      } catch (RuntimeException ignore) {
        // fix for GEODE-216
        // Most likely here would be RegionDestroyedException or CacheClosedException
        // but the cancel criteria code can throw any RuntimeException.
        // In all these cases it is ok to just fall through and return the
        // old executor type.
      }
    }
    if (forceUseOfPRExecutor) {
      return DistributionManager.PARTITIONED_REGION_EXECUTOR;
    }
    else if (ORDER_PR_GETS) {
      return DistributionManager.PARTITIONED_REGION_EXECUTOR;
    } else {
      // Make this guy serial so that it will be processed in the p2p msg reader
      // which gives it better performance.
      return DistributionManager.SERIAL_EXECUTOR;
    }
  }

  @Override
  public boolean isSevereAlertCompatible() {
    // allow forced-disconnect processing for all cache op messages
    return true;
  }
  
  @Override
  protected final boolean operateOnPartitionedRegion(
      final DistributionManager dm, PartitionedRegion r, long startTime)
      throws ForceReattemptException
  {
    if (logger.isTraceEnabled(LogMarker.DM)) {
{'fraction_non_alphanumeric': 0.10235656471599455, 'fraction_numerical': 0.0037246615571950433, 'mean_word_length': 4.529504950495049, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '22120', 'n_tokens_mistral': 3963, 'n_tokens_neox': 3811, 'n_words': 961}