qpid/java/systests/src/main/java/org/apache/qpid/test/framework/clocksynch/UDPClockSynchronizer.java - qpid - 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.qpid.test.framework.clocksynch;

 import org.apache.qpid.junit.extensions.util.CommandLineParser;
 import org.apache.qpid.junit.extensions.util.ParsedProperties;

 import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.InetAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;

 /**
  * UDPClockSynchronizer is a {@link ClockSynchronizer} that sends pings as UDP datagrams, and uses the following simple
  * algorithm to perform clock synchronization:
  *
  * <ol>
  * <li>Slave initiates synchronization with a Reference clock.</li>
  * <li>Slave stamps current local time on a "time request" message and sends to the Reference.</li>
  * <li>Upon receipt by Reference, Reference stamps Reference-time and returns.</li>
  * <li>Upon receipt by Slave, Slave subtracts current time from sent time and divides by two to compute latency. It
  *     subtracts current time from Reference time to determine Slave-Reference time delta and adds in the
  *     half-latency to get the correct clock delta.</li>
  * <li>The first result is immediately used to update the clock since it will get the local clock into at least
  *     the right ballpark.</li>
  * <li>The Slave repeats steps 2 through 4, 15 more times.</li>
  * <li>The results of the packet receipts are accumulated and sorted in lowest-latency to highest-latency order. The
  *     median latency is determined by picking the mid-point sample from this ordered list.</li>
  * <li>All samples outside 1 standard-deviation from the median are discarded and the remaining samples
  *     are averaged using an arithmetic mean.</li>
  * </ol>
  *
  * <p/>The use of UDP datagrams, instead of TCP based communication eliminates the hidden delays that TCP can introduce,
  * as it can transparently re-order or re-send packets, or introduce delays as packets are naggled.
  *
  * <p/><table id="crc"><caption>CRC Card</caption>
  * <tr><th> Responsibilities <th> Collaborations
  * <tr><td> Trigger a clock synchronziation.
  * <tr><td> Compute a clock delta to apply to the local clock.
  * <tr><td> Estimate the error in the synchronzation.
  * </table>
  */
 public class UDPClockSynchronizer implements ClockSynchronizer
 {
     /** Used for debugging. */
     // private static final Logger log = Logger.getLogger(UDPClockSynchronizer.class);

     /** Defines the timeout to use when waiting for responses to time requests. */
     private static final int TIMEOUT = 50;

     /** The clock delta. */
     private long delta = 0L;

     /** Holds an estimate of the clock error relative to the reference clock. */
     private long epsilon = 0L;

     /** Holds the address of the reference clock. */
     private InetAddress referenceAddress;

     /** Holds the socket to communicate with the reference service over. */
     private DatagramSocket socket;

     /** Used to control the shutdown in the main test loop. */
     private static boolean doSynch = true;

     /**
      * Creates a clock synchronizer against the specified address for the reference.
      *
      * @param address The address of the reference service.
      */
     public UDPClockSynchronizer(String address)
     {
         try
         {
             referenceAddress = InetAddress.getByName(address);
         }
         catch (UnknownHostException e)
         {
             throw new RuntimeException(e);
         }
     }

     /**
      * The slave side should call this to compute a clock delta with the reference.
      *
      * @throws ClockSynchFailureException If synchronization cannot be achieved, due to unavailability of the reference
      *                                    time service.
      */
     public void synch() throws ClockSynchFailureException
     {
         try
         {
             socket = new DatagramSocket();
             socket.setSoTimeout(TIMEOUT);

             // Synchronize on a single ping, to get the clock into the right ball-park.
             synch(1);

             // Synchronize on 15 pings.
             synch(15);

             // And again, for greater accuracy, on 31.
             synch(31);

             socket.close();
         }
         catch (SocketException e)
         {
             throw new RuntimeException(e);
         }
     }

     /**
      * Updates the synchronization delta by performing the specified number of reference clock requests.
      *
      * @param n The number of reference clock request cycles to perform.
      *
      * @throws ClockSynchFailureException If synchronization cannot be achieved, due to unavailability of the reference
      *                                    time service.
      */
     protected void synch(int n) throws ClockSynchFailureException
     {
         // log.debug("protected void synch(int n = " + n + "): called");

         // Create an array of deltas by performing n reference pings.
         long[] delta = new long[n];

         for (int i = 0; i < n; i++)
         {
             delta[i] = ping();
         }

         // Reject any deltas that are larger than 1 s.d. above the median.
         long median = median(delta);
         long sd = standardDeviation(delta);

         // log.debug("median = " + median);
         // log.debug("sd = " + sd);

         long[] tempDeltas = new long[n];
         int count = 0;

         for (int i = 0; i < n; i++)
         {
             if ((delta[i] <= (median + sd)) && (delta[i] >= (median - sd)))
             {
                 tempDeltas[count] = delta[i];
                 count++;
             }
             else
             {
                 // log.debug("Rejected: " + delta[i]);
             }
         }

         System.arraycopy(tempDeltas, 0, delta, 0, count);

         // Estimate the delta as the mean of the remaining deltas.
         this.delta += mean(delta);

         // Estimate the error as the standard deviation of the remaining deltas.
         this.epsilon = standardDeviation(delta);

         // log.debug("this.delta = " + this.delta);
         // log.debug("this.epsilon = " + this.epsilon);
     }

     /**
      * Performs a single reference clock request cycle and returns the estimated delta relative to the local clock.
      * This is computed as the half-latency of the requst cycle, plus the reference clock, minus the local clock.
      *
      * @return The estimated clock delta.
      *
      * @throws ClockSynchFailureException If the reference service is not responding.
      */
     protected long ping() throws ClockSynchFailureException
     {
         // log.debug("protected long ping(): called");

         try
         {
             byte[] buf = new byte[256];

             boolean timedOut = false;
             long start = 0L;
             long refTime = 0L;
             long localTime = 0L;
             long latency = 0L;
             int failCount = 0;

             // Keep trying the ping until it gets a response, or 10 tries in a row all time out.
             do
             {
                 // Start timing the request latency.
                 start = nanoTime();

                 // Get the reference time.
                 DatagramPacket packet =
                     new DatagramPacket(buf, buf.length, referenceAddress, UDPClockReference.REFERENCE_PORT);
                 socket.send(packet);
                 packet = new DatagramPacket(buf, buf.length);

                 timedOut = false;

                 try
                 {
                     socket.receive(packet);
                 }
                 catch (SocketTimeoutException e)
                 {
                     timedOut = true;
                     failCount++;

                     continue;
                 }

                 ByteBuffer bbuf = ByteBuffer.wrap(packet.getData());
                 refTime = bbuf.getLong();

                 // Stop timing the request latency.
                 localTime = nanoTime();
                 latency = localTime - start;

                 // log.debug("refTime = " + refTime);
                 // log.debug("localTime = " + localTime);
                 // log.debug("start = " + start);
                 // log.debug("latency = " + latency);
                 // log.debug("delta = " + ((latency / 2) + (refTime - localTime)));

             }
             while (timedOut && (failCount < 10));

             // Fail completely if the fail count is too high.
             if (failCount >= 10)
             {
                 throw new ClockSynchFailureException("Clock reference not responding.", null);
             }

             // Estimate delta as (ref clock + half-latency) - local clock.
             return (latency / 2) + (refTime - localTime);
         }
         catch (IOException e)
         {
             throw new RuntimeException(e);
         }
     }

     /**
      * Gets the clock delta in nano seconds.
      *
      * @return The clock delta in nano seconds.
      */
     public long getDelta()
     {
         return delta;
     }

     /**
      * Gets an estimate of the clock error in nan seconds.
      *
      * @return An estimate of the clock error in nan seconds.
      */
     public long getEpsilon()
     {
         return epsilon;
     }

     /**
      * Gets the local clock time with any computed delta added in.
      *
      * @return The local clock time with any computed delta added in.
      */
     public long nanoTime()
     {
         return System.nanoTime() + delta;
     }

     /**
      * Computes the median of a series of values.
      *
      * @param values The values.
      *
      * @return The median.
      */
     public static long median(long[] values)
     {
         // log.debug("public static long median(long[] values = " + Arrays.toString(values) + "): called");

         long median;

         // Order the list of values.
         long[] orderedValues = new long[values.length];
         System.arraycopy(values, 0, orderedValues, 0, values.length);
         Arrays.sort(orderedValues);

         // Check if the median is computed from a pair of middle value.
         if ((orderedValues.length % 2) == 0)
         {
             int middle = orderedValues.length / 2;

             median = (orderedValues[middle] + orderedValues[middle - 1]) / 2;
         }
         // The median is computed from a single middle value.
         else
         {
             median = orderedValues[orderedValues.length / 2];
         }

         // log.debug("median = " + median);

         return median;
     }

     /**
      * Computes the mean of a series of values.
      *
      * @param values The values.
      *
      * @return The mean.
      */
     public static long mean(long[] values)
     {
         // log.debug("public static long mean(long[] values = " + Arrays.toString(values) + "): called");

         long total = 0L;

         for (long value : values)
         {
             total += value;
         }

         long mean = total / values.length;

         // log.debug("mean = " + mean);

         return mean;
     }

     /**
      * Computes the variance of series of values.
      *
      * @param values The values.
      *
      * @return The variance of the values.
      */
     public static long variance(long[] values)
     {
         // log.debug("public static long variance(long[] values = " + Arrays.toString(values) + "): called");

         long mean = mean(values);

         long totalVariance = 0;

         for (long value : values)
         {
             long diff = (value - mean);
             totalVariance += diff * diff;
         }

         long variance = totalVariance / values.length;

         // log.debug("variance = " + variance);

         return variance;
     }

     /**
      * Computes the standard deviation of a series of values.
      *
      * @param values The values.
      *
      * @return The standard deviation.
      */
     public static long standardDeviation(long[] values)
     {
         // log.debug("public static long standardDeviation(long[] values = " + Arrays.toString(values) + "): called");

         long sd = Double.valueOf(Math.sqrt(variance(values))).longValue();

         // log.debug("sd = " + sd);

         return sd;
     }

     /**
      * For testing purposes. Supply address of reference clock as arg 1.
      *
      * @param args Address of reference clock as arg 1.
      */
     public static void main(String[] args)
     {
         ParsedProperties options =
             new ParsedProperties(CommandLineParser.processCommandLine(args,
                     new CommandLineParser(
                         new String[][]
                         {
                             { "1", "Address of clock reference service.", "address", "true" }
                         }), System.getProperties()));

         String address = options.getProperty("1");

         // Create a clock synchronizer.
         UDPClockSynchronizer clockSyncher = new UDPClockSynchronizer(address);

         // Set up a shutdown hook for it.
         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable()
                 {
                     public void run()
                     {
                         doSynch = false;
                     }
                 }));

         // Repeat the clock synching until the user kills the progam.
         while (doSynch)
         {
             // Perform a clock clockSynch.
             try
             {
                 clockSyncher.synch();

                 // Print out the clock delta and estimate of the error.
                 System.out.println("Delta = " + clockSyncher.getDelta());
                 System.out.println("Epsilon = " + clockSyncher.getEpsilon());

                 try
                 {
                     Thread.sleep(250);
                 }
                 catch (InterruptedException e)
                 {
                     // Restore the interrupted status and terminate the loop.
                     Thread.currentThread().interrupt();
                     doSynch = false;
                 }
             }
             // Terminate if the reference time service is unavailable.
             catch (ClockSynchFailureException e)
             {
                 doSynch = false;
             }
         }
     }
 }
	/*
	*
	* 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.qpid.test.framework.clocksynch;

	import org.apache.qpid.junit.extensions.util.CommandLineParser;
	import org.apache.qpid.junit.extensions.util.ParsedProperties;

	import java.io.IOException;
	import java.net.DatagramPacket;
	import java.net.DatagramSocket;
	import java.net.InetAddress;
	import java.net.SocketException;
	import java.net.SocketTimeoutException;
	import java.net.UnknownHostException;
	import java.nio.ByteBuffer;
	import java.util.Arrays;

	/**
	* UDPClockSynchronizer is a {@link ClockSynchronizer} that sends pings as UDP datagrams, and uses the following simple
	* algorithm to perform clock synchronization:
	*
	* <ol>
	* <li>Slave initiates synchronization with a Reference clock.</li>
	* <li>Slave stamps current local time on a "time request" message and sends to the Reference.</li>
	* <li>Upon receipt by Reference, Reference stamps Reference-time and returns.</li>
	* <li>Upon receipt by Slave, Slave subtracts current time from sent time and divides by two to compute latency. It
	* subtracts current time from Reference time to determine Slave-Reference time delta and adds in the
	* half-latency to get the correct clock delta.</li>
	* <li>The first result is immediately used to update the clock since it will get the local clock into at least
	* the right ballpark.</li>
	* <li>The Slave repeats steps 2 through 4, 15 more times.</li>
	* <li>The results of the packet receipts are accumulated and sorted in lowest-latency to highest-latency order. The
	* median latency is determined by picking the mid-point sample from this ordered list.</li>
	* <li>All samples outside 1 standard-deviation from the median are discarded and the remaining samples
	* are averaged using an arithmetic mean.</li>
	* </ol>
	*
	* <p/>The use of UDP datagrams, instead of TCP based communication eliminates the hidden delays that TCP can introduce,
	* as it can transparently re-order or re-send packets, or introduce delays as packets are naggled.
	*
	* <p/><table id="crc"><caption>CRC Card</caption>
	* <tr><th> Responsibilities <th> Collaborations
	* <tr><td> Trigger a clock synchronziation.
	* <tr><td> Compute a clock delta to apply to the local clock.
	* <tr><td> Estimate the error in the synchronzation.
	* </table>
	*/
	public class UDPClockSynchronizer implements ClockSynchronizer
	{
	/** Used for debugging. */
	// private static final Logger log = Logger.getLogger(UDPClockSynchronizer.class);

	/** Defines the timeout to use when waiting for responses to time requests. */
	private static final int TIMEOUT = 50;

	/** The clock delta. */
	private long delta = 0L;

	/** Holds an estimate of the clock error relative to the reference clock. */
	private long epsilon = 0L;

	/** Holds the address of the reference clock. */
	private InetAddress referenceAddress;

	/** Holds the socket to communicate with the reference service over. */
	private DatagramSocket socket;

	/** Used to control the shutdown in the main test loop. */
	private static boolean doSynch = true;

	/**
	* Creates a clock synchronizer against the specified address for the reference.
	*
	* @param address The address of the reference service.
	*/
	public UDPClockSynchronizer(String address)
	{
	try
	{
	referenceAddress = InetAddress.getByName(address);
	}
	catch (UnknownHostException e)
	{
	throw new RuntimeException(e);
	}
	}

	/**
	* The slave side should call this to compute a clock delta with the reference.
	*
	* @throws ClockSynchFailureException If synchronization cannot be achieved, due to unavailability of the reference
	* time service.
	*/
	public void synch() throws ClockSynchFailureException
	{
	try
	{
	socket = new DatagramSocket();
	socket.setSoTimeout(TIMEOUT);

	// Synchronize on a single ping, to get the clock into the right ball-park.
	synch(1);

	// Synchronize on 15 pings.
	synch(15);

	// And again, for greater accuracy, on 31.
	synch(31);

	socket.close();
	}
	catch (SocketException e)
	{
	throw new RuntimeException(e);
	}
	}

	/**
	* Updates the synchronization delta by performing the specified number of reference clock requests.
	*
	* @param n The number of reference clock request cycles to perform.
	*
	* @throws ClockSynchFailureException If synchronization cannot be achieved, due to unavailability of the reference
	* time service.
	*/
	protected void synch(int n) throws ClockSynchFailureException
	{
	// log.debug("protected void synch(int n = " + n + "): called");

	// Create an array of deltas by performing n reference pings.
	long[] delta = new long[n];

	for (int i = 0; i < n; i++)
	{
	delta[i] = ping();
	}

	// Reject any deltas that are larger than 1 s.d. above the median.
	long median = median(delta);
	long sd = standardDeviation(delta);

	// log.debug("median = " + median);
	// log.debug("sd = " + sd);

	long[] tempDeltas = new long[n];
	int count = 0;

	for (int i = 0; i < n; i++)
	{
	if ((delta[i] <= (median + sd)) && (delta[i] >= (median - sd)))
	{
	tempDeltas[count] = delta[i];
	count++;
	}
	else
	{
	// log.debug("Rejected: " + delta[i]);
	}
	}

	System.arraycopy(tempDeltas, 0, delta, 0, count);

	// Estimate the delta as the mean of the remaining deltas.
	this.delta += mean(delta);

	// Estimate the error as the standard deviation of the remaining deltas.
	this.epsilon = standardDeviation(delta);

	// log.debug("this.delta = " + this.delta);
	// log.debug("this.epsilon = " + this.epsilon);
	}

	/**
	* Performs a single reference clock request cycle and returns the estimated delta relative to the local clock.
	* This is computed as the half-latency of the requst cycle, plus the reference clock, minus the local clock.
	*
	* @return The estimated clock delta.
	*
	* @throws ClockSynchFailureException If the reference service is not responding.
	*/
	protected long ping() throws ClockSynchFailureException
	{
	// log.debug("protected long ping(): called");

	try
	{
	byte[] buf = new byte[256];

	boolean timedOut = false;
	long start = 0L;
	long refTime = 0L;
	long localTime = 0L;
	long latency = 0L;
	int failCount = 0;

	// Keep trying the ping until it gets a response, or 10 tries in a row all time out.
	do
	{
	// Start timing the request latency.
	start = nanoTime();

	// Get the reference time.
	DatagramPacket packet =
	new DatagramPacket(buf, buf.length, referenceAddress, UDPClockReference.REFERENCE_PORT);
	socket.send(packet);
	packet = new DatagramPacket(buf, buf.length);

	timedOut = false;

	try
	{
	socket.receive(packet);
	}
	catch (SocketTimeoutException e)
	{
	timedOut = true;
	failCount++;

	continue;
	}

	ByteBuffer bbuf = ByteBuffer.wrap(packet.getData());
	refTime = bbuf.getLong();

	// Stop timing the request latency.
	localTime = nanoTime();
	latency = localTime - start;

	// log.debug("refTime = " + refTime);
	// log.debug("localTime = " + localTime);
	// log.debug("start = " + start);
	// log.debug("latency = " + latency);
	// log.debug("delta = " + ((latency / 2) + (refTime - localTime)));

	}
	while (timedOut && (failCount < 10));

	// Fail completely if the fail count is too high.
	if (failCount >= 10)
	{
	throw new ClockSynchFailureException("Clock reference not responding.", null);
	}

	// Estimate delta as (ref clock + half-latency) - local clock.
	return (latency / 2) + (refTime - localTime);
	}
	catch (IOException e)
	{
	throw new RuntimeException(e);
	}
	}

	/**
	* Gets the clock delta in nano seconds.
	*
	* @return The clock delta in nano seconds.
	*/
	public long getDelta()
	{
	return delta;
	}

	/**
	* Gets an estimate of the clock error in nan seconds.
	*
	* @return An estimate of the clock error in nan seconds.
	*/
	public long getEpsilon()
	{
	return epsilon;
	}

	/**
	* Gets the local clock time with any computed delta added in.
	*
	* @return The local clock time with any computed delta added in.
	*/
	public long nanoTime()
	{
	return System.nanoTime() + delta;
	}

	/**
	* Computes the median of a series of values.
	*
	* @param values The values.
	*
	* @return The median.
	*/
	public static long median(long[] values)
	{
	// log.debug("public static long median(long[] values = " + Arrays.toString(values) + "): called");

	long median;

	// Order the list of values.
	long[] orderedValues = new long[values.length];
	System.arraycopy(values, 0, orderedValues, 0, values.length);
	Arrays.sort(orderedValues);

	// Check if the median is computed from a pair of middle value.
	if ((orderedValues.length % 2) == 0)
	{
	int middle = orderedValues.length / 2;

	median = (orderedValues[middle] + orderedValues[middle - 1]) / 2;
	}
	// The median is computed from a single middle value.
	else
	{
	median = orderedValues[orderedValues.length / 2];
	}

	// log.debug("median = " + median);

	return median;
	}

	/**
	* Computes the mean of a series of values.
	*
	* @param values The values.
	*
	* @return The mean.
	*/
	public static long mean(long[] values)
	{
	// log.debug("public static long mean(long[] values = " + Arrays.toString(values) + "): called");

	long total = 0L;

	for (long value : values)
	{
	total += value;
	}

	long mean = total / values.length;

	// log.debug("mean = " + mean);

	return mean;
	}

	/**
	* Computes the variance of series of values.
	*
	* @param values The values.
	*
	* @return The variance of the values.
	*/
	public static long variance(long[] values)
	{
	// log.debug("public static long variance(long[] values = " + Arrays.toString(values) + "): called");

	long mean = mean(values);

	long totalVariance = 0;

	for (long value : values)
	{
	long diff = (value - mean);
	totalVariance += diff * diff;
	}

	long variance = totalVariance / values.length;

	// log.debug("variance = " + variance);

	return variance;
	}

	/**
	* Computes the standard deviation of a series of values.
	*
	* @param values The values.
	*
	* @return The standard deviation.
	*/
	public static long standardDeviation(long[] values)
	{
	// log.debug("public static long standardDeviation(long[] values = " + Arrays.toString(values) + "): called");

	long sd = Double.valueOf(Math.sqrt(variance(values))).longValue();

	// log.debug("sd = " + sd);

	return sd;
	}

	/**
	* For testing purposes. Supply address of reference clock as arg 1.
	*
	* @param args Address of reference clock as arg 1.
	*/
	public static void main(String[] args)
	{
	ParsedProperties options =
	new ParsedProperties(CommandLineParser.processCommandLine(args,
	new CommandLineParser(
	new String[][]
	{
	{ "1", "Address of clock reference service.", "address", "true" }
	}), System.getProperties()));

	String address = options.getProperty("1");

	// Create a clock synchronizer.
	UDPClockSynchronizer clockSyncher = new UDPClockSynchronizer(address);

	// Set up a shutdown hook for it.
	Runtime.getRuntime().addShutdownHook(new Thread(new Runnable()
	{
	public void run()
	{
	doSynch = false;
	}
	}));

	// Repeat the clock synching until the user kills the progam.
	while (doSynch)
	{
	// Perform a clock clockSynch.
	try
	{
	clockSyncher.synch();

	// Print out the clock delta and estimate of the error.
	System.out.println("Delta = " + clockSyncher.getDelta());
	System.out.println("Epsilon = " + clockSyncher.getEpsilon());

	try
	{
	Thread.sleep(250);
	}
	catch (InterruptedException e)
	{
	// Restore the interrupted status and terminate the loop.
	Thread.currentThread().interrupt();
	doSynch = false;
	}
	}
	// Terminate if the reference time service is unavailable.
	catch (ClockSynchFailureException e)
	{
	doSynch = false;
	}
	}
	}
	}