geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IdentityRequestMessage.java - geode - Git at Google

 /*
  * 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.partitioned;

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

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.SystemFailure;
 import org.apache.geode.annotations.internal.MakeNotStatic;
 import org.apache.geode.distributed.internal.ClusterDistributionManager;
 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.HighPriorityDistributionMessage;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.distributed.internal.MessageWithReply;
 import org.apache.geode.distributed.internal.OperationExecutors;
 import org.apache.geode.distributed.internal.ReplyException;
 import org.apache.geode.distributed.internal.ReplyProcessor21;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.Assert;
 import org.apache.geode.internal.logging.log4j.LogMarker;
 import org.apache.geode.internal.serialization.DeserializationContext;
 import org.apache.geode.internal.serialization.SerializationContext;
 import org.apache.geode.logging.internal.log4j.api.LogService;

 /**
  * A message sent to determine the most recent PartitionedRegion identity
  *
  * @since GemFire 5.0
  */
 public class IdentityRequestMessage extends DistributionMessage implements MessageWithReply {
   private static final Logger logger = LogService.getLogger();

   private static final int UNINITIALIZED = -1;

   /**
    * This is the keeper of the latest PartitionedRegion Identity on a per VM basis
    */
   @MakeNotStatic
   private static int latestId = UNINITIALIZED;

   public static synchronized void setLatestId(int newlatest) {
     if (newlatest > latestId) {
       latestId = newlatest;
     }
   }

   /**
    * Method public for test reasons
    *
    * @return the latest identity
    */
   public static synchronized int getLatestId() {
     return latestId;
   }

   private int processorId;

   /**
    * Empty constructor to conform to DataSerializer interface
    *
    */
   public IdentityRequestMessage() {}

   public IdentityRequestMessage(Set recipients, int processorId) {
     setRecipients(recipients);
     this.processorId = processorId;
   }

   @Override
   protected void process(ClusterDistributionManager dm) {
     try {
       if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
         logger.trace(LogMarker.DM_VERBOSE, "{}: processing message {}", getClass().getName(), this);
       }

       IdentityReplyMessage.send(getSender(), getProcessorId(), dm);
     } catch (VirtualMachineError err) {
       SystemFailure.initiateFailure(err);
       // If this ever returns, rethrow the error. We're poisoned
       // now, so don't let this thread continue.
       throw err;
     } catch (Throwable t) {
       // Whenever you catch Error or Throwable, you must also
       // catch VirtualMachineError (see above). However, there is
       // _still_ a possibility that you are dealing with a cascading
       // error condition, so you also need to check to see if the JVM
       // is still usable:
       SystemFailure.checkFailure();
       logger.debug("{} Caught throwable {}", this, t.getMessage(), t);
     }
   }

   @Override
   public int getProcessorId() {
     return this.processorId;
   }


   @Override
   public int getProcessorType() {
     return OperationExecutors.HIGH_PRIORITY_EXECUTOR;
   }

   /**
    * Sends a <code>IdentityRequest</code> to each <code>PartitionedRegion</code>
    * {@link org.apache.geode.internal.cache.Node}. The <code>IdentityResponse</code> is used to
    * fetch the highest current identity value.
    *
    * @return the response object to wait upon
    */
   public static IdentityResponse send(Set recipients, InternalDistributedSystem is) {
     Assert.assertTrue(recipients != null, "IdentityMessage NULL recipients set");
     int i = 0;
     for (Iterator ri = recipients.iterator(); ri.hasNext(); i++) {
       Assert.assertTrue(null != ri.next(), "IdenityMessage recipient " + i + " is null");
     }

     IdentityResponse p = new IdentityResponse(is, recipients);
     IdentityRequestMessage m = new IdentityRequestMessage(recipients, p.getProcessorId());
     is.getDistributionManager().putOutgoing(m);
     return p;
   }


   @Override
   public int getDSFID() {
     return PR_IDENTITY_REQUEST_MESSAGE;
   }

   @Override
   public void fromData(DataInput in,
       DeserializationContext context) throws IOException, ClassNotFoundException {
     super.fromData(in, context);
     this.processorId = in.readInt();
   }

   @Override
   public void toData(DataOutput out,
       SerializationContext context) throws IOException {
     super.toData(out, context);
     out.writeInt(this.processorId);
   }

   @Override
   public String toString() {
     return new StringBuffer().append(getClass().getName()).append("(sender=").append(getSender())
         .append("; processorId=").append(this.processorId).append(")").toString();
   }


   /**
    * The message that contains the <code>Integer</code> identity response to the
    * {@link IdentityRequestMessage}
    *
    * @since GemFire 5.0
    */
   public static class IdentityReplyMessage extends HighPriorityDistributionMessage {
     private int Id = UNINITIALIZED;

     /** The shared obj id of the ReplyProcessor */
     private int processorId;

     /**
      * Empty constructor to conform to DataSerializable interface
      */
     public IdentityReplyMessage() {}

     private IdentityReplyMessage(int processorId) {
       this.processorId = processorId;
       this.Id = IdentityRequestMessage.getLatestId();
     }

     public static void send(InternalDistributedMember recipient, int processorId,
         DistributionManager dm) {
       Assert.assertTrue(recipient != null, "IdentityReplyMessage NULL reply message");
       IdentityReplyMessage m = new IdentityReplyMessage(processorId);
       m.setRecipient(recipient);
       dm.putOutgoing(m);
     }

     @Override
     protected void process(final ClusterDistributionManager dm) {
       final long startTime = getTimestamp();
       if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
         logger.trace(LogMarker.DM_VERBOSE,
             "{} process invoking reply processor with processorId:{}", getClass().getName(),
             this.processorId);
       }

       ReplyProcessor21 processor = ReplyProcessor21.getProcessor(this.processorId);

       if (processor == null) {
         if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
           logger.trace(LogMarker.DM_VERBOSE, "Processor not found: {}", getClass().getName());
         }
         return;
       }
       processor.process(this);

       if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
         logger.trace(LogMarker.DM_VERBOSE, "{} Processed {}", processor, this);
       }
       dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
     }

     @Override
     public void toData(DataOutput out,
         SerializationContext context) throws IOException {
       super.toData(out, context);
       out.writeInt(this.processorId);
       out.writeInt(this.Id);
     }

     @Override
     public int getDSFID() {
       return PR_IDENTITY_REPLY_MESSAGE;
     }

     @Override
     public void fromData(DataInput in,
         DeserializationContext context) throws IOException, ClassNotFoundException {
       super.fromData(in, context);
       this.processorId = in.readInt();
       this.Id = in.readInt();
     }

     @Override
     public String toString() {
       return new StringBuffer().append(getClass().getName()).append("(sender=").append(getSender())
           .append("; processorId=").append(this.processorId).append("; PRId=").append(getId())
           .append(")").toString();
     }

     /**
      * Fetch the current Identity number
      *
      * @return the identity Integer from the sender or null if the sender did not have the Integer
      *         initialized
      */
     public Integer getId() {
       if (this.Id == UNINITIALIZED) {
         return null;
       }
       return Integer.valueOf(this.Id);
     }
   }

   /**
    * The response to a {@link IdentityRequestMessage} use {@link #waitForId()} to capture the
    * identity
    *
    * @since GemFire 5.0
    */
   public static class IdentityResponse extends ReplyProcessor21 {
     private Integer returnValue;

     public IdentityResponse(InternalDistributedSystem system, Set initMembers) {
       super(system, initMembers);
       int localIdent = IdentityRequestMessage.getLatestId();
       if (localIdent != UNINITIALIZED) {
         this.returnValue = Integer.valueOf(localIdent);
       }
     }

     @Override
     public void process(DistributionMessage msg) {
       try {
         if (msg instanceof IdentityReplyMessage) {
           IdentityReplyMessage reply = (IdentityReplyMessage) msg;
           final Integer remoteId = reply.getId();
           synchronized (this) {
             if (remoteId != null) {
               if (this.returnValue == null) {
                 this.returnValue = remoteId;
               } else {
                 if (remoteId.intValue() > this.returnValue.intValue()) {
                   this.returnValue = remoteId;
                 }
               }
             }
           }
           if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
             logger.trace(LogMarker.DM_VERBOSE, "{} return value is {}", getClass().getName(),
                 this.returnValue);
           }
         }
       } finally {
         super.process(msg);
       }
     }

     /**
      * Fetch the next <code>PartitionedRegion</code> identity, used to uniquely identify (globally)
      * each instance of a <code>PartitionedRegion</code>
      *
      * @return the next highest Integer for the <code>PartitionedRegion</code> or null if this is
      *         the first identity
      *
      * @see PartitionMessage#getRegionId()
      */
     public Integer waitForId() {
       try {
         waitForRepliesUninterruptibly();
       } catch (ReplyException e) {
         logger.debug("{} waitBucketSizes ignoring exception {}", getClass().getName(),
             e.getMessage(), e);
       }
       synchronized (this) {
         return this.returnValue;
       }
     }
   }

 }
	/*
	* 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.partitioned;

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

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.SystemFailure;
	import org.apache.geode.annotations.internal.MakeNotStatic;
	import org.apache.geode.distributed.internal.ClusterDistributionManager;
	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.HighPriorityDistributionMessage;
	import org.apache.geode.distributed.internal.InternalDistributedSystem;
	import org.apache.geode.distributed.internal.MessageWithReply;
	import org.apache.geode.distributed.internal.OperationExecutors;
	import org.apache.geode.distributed.internal.ReplyException;
	import org.apache.geode.distributed.internal.ReplyProcessor21;
	import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
	import org.apache.geode.internal.Assert;
	import org.apache.geode.internal.logging.log4j.LogMarker;
	import org.apache.geode.internal.serialization.DeserializationContext;
	import org.apache.geode.internal.serialization.SerializationContext;
	import org.apache.geode.logging.internal.log4j.api.LogService;

	/**
	* A message sent to determine the most recent PartitionedRegion identity
	*
	* @since GemFire 5.0
	*/
	public class IdentityRequestMessage extends DistributionMessage implements MessageWithReply {
	private static final Logger logger = LogService.getLogger();

	private static final int UNINITIALIZED = -1;

	/**
	* This is the keeper of the latest PartitionedRegion Identity on a per VM basis
	*/
	@MakeNotStatic
	private static int latestId = UNINITIALIZED;

	public static synchronized void setLatestId(int newlatest) {
	if (newlatest > latestId) {
	latestId = newlatest;
	}
	}

	/**
	* Method public for test reasons
	*
	* @return the latest identity
	*/
	public static synchronized int getLatestId() {
	return latestId;
	}

	private int processorId;

	/**
	* Empty constructor to conform to DataSerializer interface
	*
	*/
	public IdentityRequestMessage() {}

	public IdentityRequestMessage(Set recipients, int processorId) {
	setRecipients(recipients);
	this.processorId = processorId;
	}

	@Override
	protected void process(ClusterDistributionManager dm) {
	try {
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE, "{}: processing message {}", getClass().getName(), this);
	}

	IdentityReplyMessage.send(getSender(), getProcessorId(), dm);
	} catch (VirtualMachineError err) {
	SystemFailure.initiateFailure(err);
	// If this ever returns, rethrow the error. We're poisoned
	// now, so don't let this thread continue.
	throw err;
	} catch (Throwable t) {
	// Whenever you catch Error or Throwable, you must also
	// catch VirtualMachineError (see above). However, there is
	// _still_ a possibility that you are dealing with a cascading
	// error condition, so you also need to check to see if the JVM
	// is still usable:
	SystemFailure.checkFailure();
	logger.debug("{} Caught throwable {}", this, t.getMessage(), t);
	}
	}

	@Override
	public int getProcessorId() {
	return this.processorId;
	}


	@Override
	public int getProcessorType() {
	return OperationExecutors.HIGH_PRIORITY_EXECUTOR;
	}

	/**
	* Sends a <code>IdentityRequest</code> to each <code>PartitionedRegion</code>
	* {@link org.apache.geode.internal.cache.Node}. The <code>IdentityResponse</code> is used to
	* fetch the highest current identity value.
	*
	* @return the response object to wait upon
	*/
	public static IdentityResponse send(Set recipients, InternalDistributedSystem is) {
	Assert.assertTrue(recipients != null, "IdentityMessage NULL recipients set");
	int i = 0;
	for (Iterator ri = recipients.iterator(); ri.hasNext(); i++) {
	Assert.assertTrue(null != ri.next(), "IdenityMessage recipient " + i + " is null");
	}

	IdentityResponse p = new IdentityResponse(is, recipients);
	IdentityRequestMessage m = new IdentityRequestMessage(recipients, p.getProcessorId());
	is.getDistributionManager().putOutgoing(m);
	return p;
	}


	@Override
	public int getDSFID() {
	return PR_IDENTITY_REQUEST_MESSAGE;
	}

	@Override
	public void fromData(DataInput in,
	DeserializationContext context) throws IOException, ClassNotFoundException {
	super.fromData(in, context);
	this.processorId = in.readInt();
	}

	@Override
	public void toData(DataOutput out,
	SerializationContext context) throws IOException {
	super.toData(out, context);
	out.writeInt(this.processorId);
	}

	@Override
	public String toString() {
	return new StringBuffer().append(getClass().getName()).append("(sender=").append(getSender())
	.append("; processorId=").append(this.processorId).append(")").toString();
	}


	/**
	* The message that contains the <code>Integer</code> identity response to the
	* {@link IdentityRequestMessage}
	*
	* @since GemFire 5.0
	*/
	public static class IdentityReplyMessage extends HighPriorityDistributionMessage {
	private int Id = UNINITIALIZED;

	/** The shared obj id of the ReplyProcessor */
	private int processorId;

	/**
	* Empty constructor to conform to DataSerializable interface
	*/
	public IdentityReplyMessage() {}

	private IdentityReplyMessage(int processorId) {
	this.processorId = processorId;
	this.Id = IdentityRequestMessage.getLatestId();
	}

	public static void send(InternalDistributedMember recipient, int processorId,
	DistributionManager dm) {
	Assert.assertTrue(recipient != null, "IdentityReplyMessage NULL reply message");
	IdentityReplyMessage m = new IdentityReplyMessage(processorId);
	m.setRecipient(recipient);
	dm.putOutgoing(m);
	}

	@Override
	protected void process(final ClusterDistributionManager dm) {
	final long startTime = getTimestamp();
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE,
	"{} process invoking reply processor with processorId:{}", getClass().getName(),
	this.processorId);
	}

	ReplyProcessor21 processor = ReplyProcessor21.getProcessor(this.processorId);

	if (processor == null) {
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE, "Processor not found: {}", getClass().getName());
	}
	return;
	}
	processor.process(this);

	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE, "{} Processed {}", processor, this);
	}
	dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
	}

	@Override
	public void toData(DataOutput out,
	SerializationContext context) throws IOException {
	super.toData(out, context);
	out.writeInt(this.processorId);
	out.writeInt(this.Id);
	}

	@Override
	public int getDSFID() {
	return PR_IDENTITY_REPLY_MESSAGE;
	}

	@Override
	public void fromData(DataInput in,
	DeserializationContext context) throws IOException, ClassNotFoundException {
	super.fromData(in, context);
	this.processorId = in.readInt();
	this.Id = in.readInt();
	}

	@Override
	public String toString() {
	return new StringBuffer().append(getClass().getName()).append("(sender=").append(getSender())
	.append("; processorId=").append(this.processorId).append("; PRId=").append(getId())
	.append(")").toString();
	}

	/**
	* Fetch the current Identity number
	*
	* @return the identity Integer from the sender or null if the sender did not have the Integer
	* initialized
	*/
	public Integer getId() {
	if (this.Id == UNINITIALIZED) {
	return null;
	}
	return Integer.valueOf(this.Id);
	}
	}

	/**
	* The response to a {@link IdentityRequestMessage} use {@link #waitForId()} to capture the
	* identity
	*
	* @since GemFire 5.0
	*/
	public static class IdentityResponse extends ReplyProcessor21 {
	private Integer returnValue;

	public IdentityResponse(InternalDistributedSystem system, Set initMembers) {
	super(system, initMembers);
	int localIdent = IdentityRequestMessage.getLatestId();
	if (localIdent != UNINITIALIZED) {
	this.returnValue = Integer.valueOf(localIdent);
	}
	}

	@Override
	public void process(DistributionMessage msg) {
	try {
	if (msg instanceof IdentityReplyMessage) {
	IdentityReplyMessage reply = (IdentityReplyMessage) msg;
	final Integer remoteId = reply.getId();
	synchronized (this) {
	if (remoteId != null) {
	if (this.returnValue == null) {
	this.returnValue = remoteId;
	} else {
	if (remoteId.intValue() > this.returnValue.intValue()) {
	this.returnValue = remoteId;
	}
	}
	}
	}
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE, "{} return value is {}", getClass().getName(),
	this.returnValue);
	}
	}
	} finally {
	super.process(msg);
	}
	}

	/**
	* Fetch the next <code>PartitionedRegion</code> identity, used to uniquely identify (globally)
	* each instance of a <code>PartitionedRegion</code>
	*
	* @return the next highest Integer for the <code>PartitionedRegion</code> or null if this is
	* the first identity
	*
	* @see PartitionMessage#getRegionId()
	*/
	public Integer waitForId() {
	try {
	waitForRepliesUninterruptibly();
	} catch (ReplyException e) {
	logger.debug("{} waitBucketSizes ignoring exception {}", getClass().getName(),
	e.getMessage(), e);
	}
	synchronized (this) {
	return this.returnValue;
	}
	}
	}

	}