configadmin/src/test/java/org/apache/felix/cm/integration/ConfigurationAdminUpdateStressTest.java - felix - 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.felix.cm.integration;


 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.Dictionary;
 import java.util.HashSet;
 import java.util.Hashtable;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;

 import org.junit.Assert;

 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.ops4j.pax.exam.junit.JUnit4TestRunner;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.ServiceReference;
 import org.osgi.service.cm.Configuration;
 import org.osgi.service.cm.ConfigurationAdmin;
 import org.osgi.service.cm.ConfigurationException;
 import org.osgi.service.cm.ManagedServiceFactory;
 import org.osgi.service.log.LogService;
 import org.osgi.util.tracker.ServiceTracker;


 /**
  * The <code>ConfigurationAdminUpdateStressTest</code> repeatedly updates
  * a ManagedFactoryService with configuration to verify configuration is
  * exactly delivered once and no update is lost.
  *
  * @see <a href="https://issues.apache.org/jira/browse/FELIX-1545">FELIX-1545</a>
  */
 @RunWith(JUnit4TestRunner.class)
 public class ConfigurationAdminUpdateStressTest extends ConfigurationTestBase implements LogService
 {
     public static final int TEST_LOOP = 10;
     public static final int UPDATE_LOOP = 100;

     private String _FACTORYPID = "MyPID";
     private String _FACTORYNAME = "MyName";

     private volatile CountDownLatch _factoryConfigCreateLatch;
     private volatile CountDownLatch _factoryConfigUpdateLatch;
     private volatile CountDownLatch _factoryConfigDeleteLatch;
     private volatile CountDownLatch _testLatch;
     private volatile ServiceTracker<ConfigurationAdmin, ConfigurationAdmin> _tracker;


     // ----------------------- Initialization -------------------------------------------

     @Before
     public void startup()
     {
         bundleContext.registerService( LogService.class.getName(), this, null );
         _tracker = new ServiceTracker<>( bundleContext, ConfigurationAdmin.class, null );
         _tracker.open();
     }


     /**
      * Always cleanup our bundle location file (because pax seems to forget to cleanup it)
      * @param context
      */

     @After
     public void tearDown()
     {
         _tracker.close();
     }


     // ----------------------- LogService -----------------------------------------------

     public void log( int level, String message )
     {
         System.out.println( "[LogService/" + level + "] " + message );
     }


     public void log( int level, String message, Throwable exception )
     {
         StringBuilder sb = new StringBuilder();
         sb.append( "[LogService/" + level + "] " );
         sb.append( message );
         parse( sb, exception );
         System.out.println( sb.toString() );
     }


     public void log( @SuppressWarnings("rawtypes") ServiceReference sr, int level, String message )
     {
         StringBuilder sb = new StringBuilder();
         sb.append( "[LogService/" + level + "] " );
         sb.append( message );
         System.out.println( sb.toString() );
     }


     public void log( @SuppressWarnings("rawtypes") ServiceReference sr, int level, String message, Throwable exception )
     {
         StringBuilder sb = new StringBuilder();
         sb.append( "[LogService/" + level + "] " );
         sb.append( message );
         parse( sb, exception );
         System.out.println( sb.toString() );
     }


     private void parse( StringBuilder sb, Throwable t )
     {
         if ( t != null )
         {
             sb.append( " - " );
             StringWriter buffer = new StringWriter();
             PrintWriter pw = new PrintWriter( buffer );
             t.printStackTrace( pw );
             sb.append( buffer.toString() );
         }
     }


     // --------------------------- CM Update stress test -------------------------------------

     @Test
     public void testCMUpdateStress()
     {
         _testLatch = new CountDownLatch( 1 );
         try
         {
             CreateStress stress = new CreateUpdateStress( bundleContext );
             stress.start();

             if ( !_testLatch.await( 15, TimeUnit.SECONDS ) )
             {

                 log( LogService.LOG_DEBUG, "create latch: " + _factoryConfigCreateLatch.getCount() );
                 log( LogService.LOG_DEBUG, "update latch: " + _factoryConfigUpdateLatch.getCount() );
                 log( LogService.LOG_DEBUG, "delete latch: " + _factoryConfigDeleteLatch.getCount() );

                 Assert.fail( "Test did not completed timely" );
             }
         }
         catch ( InterruptedException e )
         {
             Assert.fail( "Test interrupted" );
         }
     }

     @Test
     public void testCMUpdateNamedStress()
     {
     	_testLatch = new CountDownLatch( 1 );
     	try
     	{
     		CreateStress stress = new CreateUpdateNamedStress( bundleContext );
     		stress.start();

     		if ( !_testLatch.await( 15, TimeUnit.SECONDS ) )
     		{

     			log( LogService.LOG_DEBUG, "create latch: " + _factoryConfigCreateLatch.getCount() );
     			log( LogService.LOG_DEBUG, "update latch: " + _factoryConfigUpdateLatch.getCount() );
     			log( LogService.LOG_DEBUG, "delete latch: " + _factoryConfigDeleteLatch.getCount() );

     			Assert.fail( "Test did not completed timely" );
     		}
     	}
     	catch ( InterruptedException e )
     	{
     		Assert.fail( "Test interrupted" );
     	}
     }


     /**
      * Setup the latches used throughout this test
      */
     private void setupLatches()
     {
         _factoryConfigCreateLatch = new CountDownLatch( 1 );
         _factoryConfigUpdateLatch = new CountDownLatch( UPDATE_LOOP );
         _factoryConfigDeleteLatch = new CountDownLatch( 1 );
     }

     /**
      * This is our Factory class which will react up CM factory configuration objects.
      * Each time a factory configuration object is created, the _factoryConfigCreatedLatch is counted down.
      * Each time a factory configuration object is updated, the _factoryConfigUpdatedLatch is counted down.
      * Each time a factory configuration object is deleted, the _factoryConfigDeletedLatch is counted down.
      */
     @Ignore
     class Factory implements ManagedServiceFactory
     {
         Set<String> _pids = new HashSet<String>();


         public synchronized void updated( String pid, Dictionary<String, ?> properties ) throws ConfigurationException
         {
             if ( _pids.add( pid ) )
             {
                 // pid created
                 _factoryConfigCreateLatch.countDown();
                 log( LogService.LOG_DEBUG, "Config created; create latch= " + _factoryConfigCreateLatch.getCount() );
             }
             else
             {
                 // pid updated
                 try
                 {
                     Long number = ( Long ) properties.get( "number" );
                     long currentNumber = _factoryConfigUpdateLatch.getCount();
                     if ( number.longValue() != currentNumber )
                     {
                         throw new ConfigurationException( "number", "Expected number=" + currentNumber + ", actual="
                             + number );
                     }
                     _factoryConfigUpdateLatch.countDown();
                     log( LogService.LOG_DEBUG, "Config updated; update latch= " + _factoryConfigUpdateLatch.getCount()
                         + " (number=" + number + ")" );
                 }
                 catch ( ClassCastException e )
                 {
                     throw new ConfigurationException( "number", e.getMessage(), e );
                 }
             }
         }


         public void deleted( String pid )
         {
         	// We need to remove this as the same PID can re-occur after deletion
         	_pids.remove(pid);
             _factoryConfigDeleteLatch.countDown();
             log( LogService.LOG_DEBUG, "Config deleted; delete latch= " + _factoryConfigDeleteLatch.getCount() );
         }


         public String getName()
         {
             return "MyPID";
         }
     }

     /**
      * This class creates/update/delete some factory configuration instances, using a separate thread.
      */
     @Ignore
     abstract class CreateStress extends Thread
     {
         BundleContext _bc;


         CreateStress( BundleContext bctx )
         {
             _bc = bctx;
         }


         public void run()
         {
             try
             {
                 System.out.println( "Starting CM stress test ..." );
                 ConfigurationAdmin cm = ( ConfigurationAdmin ) _tracker.waitForService( 2000 );
                 setupLatches();
                 Factory factory = new Factory();
                 Hashtable<String, Object> serviceProps = new Hashtable<String, Object>();
                 serviceProps.put( "service.pid", _FACTORYPID );
                 _bc.registerService( ManagedServiceFactory.class.getName(), factory, serviceProps );

                 for ( int l = 0; l < TEST_LOOP; l++ )
                 {
                     // Create factory configuration
                     Hashtable<String, Object> props = new Hashtable<String, Object>();
                     props.put( "foo", "bar" );
                     obtainConfiguration(cm).update( props );

                     // Check if our Factory has seen the factory configuration creation
                     if ( !_factoryConfigCreateLatch.await( 10, TimeUnit.SECONDS ) )
                     {
                         throw new RuntimeException( "_factoryConfigCreateLatch did not reach zero timely" );
                     }

                     // Update factory configuration many times
                     for ( int i = 0; i < UPDATE_LOOP; i++ )
                     {
                         props = new Hashtable<String, Object>();
                         props.put( "foo", "bar" + i );
                         props.put( "number", Long.valueOf( UPDATE_LOOP - i ) );
                         obtainConfiguration(cm).update( props );
                     }

                     // Check if all configuration updates have been caught by our Factory
                     if ( !_factoryConfigUpdateLatch.await( 10, TimeUnit.SECONDS ) )
                     {
                         throw new RuntimeException( "_factoryConfigUpdateLatch did not reach zero timely" );
                     }

                     // Remove factory configuration
                     obtainConfiguration(cm).delete();

                     // Check if our Factory has seen the configration removal
                     if ( !_factoryConfigDeleteLatch.await( 10, TimeUnit.SECONDS ) )
                     {
                         throw new RuntimeException( "_factoryConfigDeleteLatch did not reach zero timely" );
                     }

                     // Reset latches
                     setupLatches();
                 }
             }
             catch ( Exception e )
             {
                 e.printStackTrace( System.err );
                 return;
             }
             _testLatch.countDown(); // Notify that our test is done
         }


 		protected abstract org.osgi.service.cm.Configuration obtainConfiguration(ConfigurationAdmin cm)
 				throws Exception;

     }

     @Ignore
     class CreateUpdateStress extends CreateStress {
     	CreateUpdateStress(BundleContext bctx) {
 			super(bctx);
 		}

 		protected org.osgi.service.cm.Configuration obtainConfiguration(ConfigurationAdmin cm)
 				throws Exception {
 			Configuration[] cfgs = cm.listConfigurations("(service.factoryPid=" + _FACTORYPID + ")" );

 			org.osgi.service.cm.Configuration conf;
 			if(cfgs == null) {
 				conf = cm.createFactoryConfiguration( _FACTORYPID, null );
 			} else if (cfgs.length == 1) {
 				conf = cfgs[0];
 			} else {
 				throw new IllegalArgumentException("Only one configuration expected");
 			}

 			return conf;
 		}
     }

     @Ignore
     class CreateUpdateNamedStress extends CreateStress {
     	CreateUpdateNamedStress(BundleContext bctx) {
     		super(bctx);
     	}

 		protected org.osgi.service.cm.Configuration obtainConfiguration(ConfigurationAdmin cm)
 				throws IOException {
 			org.osgi.service.cm.Configuration conf = cm.getFactoryConfiguration( _FACTORYPID, _FACTORYNAME, null );
 			return conf;
 		}
     }
 }
	/*
	* 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.felix.cm.integration;


	import java.io.IOException;
	import java.io.PrintWriter;
	import java.io.StringWriter;
	import java.util.Dictionary;
	import java.util.HashSet;
	import java.util.Hashtable;
	import java.util.Set;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.TimeUnit;

	import org.junit.Assert;

	import org.junit.After;
	import org.junit.Before;
	import org.junit.Ignore;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.ops4j.pax.exam.junit.JUnit4TestRunner;
	import org.osgi.framework.BundleContext;
	import org.osgi.framework.ServiceReference;
	import org.osgi.service.cm.Configuration;
	import org.osgi.service.cm.ConfigurationAdmin;
	import org.osgi.service.cm.ConfigurationException;
	import org.osgi.service.cm.ManagedServiceFactory;
	import org.osgi.service.log.LogService;
	import org.osgi.util.tracker.ServiceTracker;


	/**
	* The <code>ConfigurationAdminUpdateStressTest</code> repeatedly updates
	* a ManagedFactoryService with configuration to verify configuration is
	* exactly delivered once and no update is lost.
	*
	* @see <a href="https://issues.apache.org/jira/browse/FELIX-1545">FELIX-1545</a>
	*/
	@RunWith(JUnit4TestRunner.class)
	public class ConfigurationAdminUpdateStressTest extends ConfigurationTestBase implements LogService
	{
	public static final int TEST_LOOP = 10;
	public static final int UPDATE_LOOP = 100;

	private String _FACTORYPID = "MyPID";
	private String _FACTORYNAME = "MyName";

	private volatile CountDownLatch _factoryConfigCreateLatch;
	private volatile CountDownLatch _factoryConfigUpdateLatch;
	private volatile CountDownLatch _factoryConfigDeleteLatch;
	private volatile CountDownLatch _testLatch;
	private volatile ServiceTracker<ConfigurationAdmin, ConfigurationAdmin> _tracker;


	// ----------------------- Initialization -------------------------------------------

	@Before
	public void startup()
	{
	bundleContext.registerService( LogService.class.getName(), this, null );
	_tracker = new ServiceTracker<>( bundleContext, ConfigurationAdmin.class, null );
	_tracker.open();
	}


	/**
	* Always cleanup our bundle location file (because pax seems to forget to cleanup it)
	* @param context
	*/

	@After
	public void tearDown()
	{
	_tracker.close();
	}


	// ----------------------- LogService -----------------------------------------------

	public void log( int level, String message )
	{
	System.out.println( "[LogService/" + level + "] " + message );
	}


	public void log( int level, String message, Throwable exception )
	{
	StringBuilder sb = new StringBuilder();
	sb.append( "[LogService/" + level + "] " );
	sb.append( message );
	parse( sb, exception );
	System.out.println( sb.toString() );
	}


	public void log( @SuppressWarnings("rawtypes") ServiceReference sr, int level, String message )
	{
	StringBuilder sb = new StringBuilder();
	sb.append( "[LogService/" + level + "] " );
	sb.append( message );
	System.out.println( sb.toString() );
	}


	public void log( @SuppressWarnings("rawtypes") ServiceReference sr, int level, String message, Throwable exception )
	{
	StringBuilder sb = new StringBuilder();
	sb.append( "[LogService/" + level + "] " );
	sb.append( message );
	parse( sb, exception );
	System.out.println( sb.toString() );
	}


	private void parse( StringBuilder sb, Throwable t )
	{
	if ( t != null )
	{
	sb.append( " - " );
	StringWriter buffer = new StringWriter();
	PrintWriter pw = new PrintWriter( buffer );
	t.printStackTrace( pw );
	sb.append( buffer.toString() );
	}
	}


	// --------------------------- CM Update stress test -------------------------------------

	@Test
	public void testCMUpdateStress()
	{
	_testLatch = new CountDownLatch( 1 );
	try
	{
	CreateStress stress = new CreateUpdateStress( bundleContext );
	stress.start();

	if ( !_testLatch.await( 15, TimeUnit.SECONDS ) )
	{

	log( LogService.LOG_DEBUG, "create latch: " + _factoryConfigCreateLatch.getCount() );
	log( LogService.LOG_DEBUG, "update latch: " + _factoryConfigUpdateLatch.getCount() );
	log( LogService.LOG_DEBUG, "delete latch: " + _factoryConfigDeleteLatch.getCount() );

	Assert.fail( "Test did not completed timely" );
	}
	}
	catch ( InterruptedException e )
	{
	Assert.fail( "Test interrupted" );
	}
	}

	@Test
	public void testCMUpdateNamedStress()
	{
	_testLatch = new CountDownLatch( 1 );
	try
	{
	CreateStress stress = new CreateUpdateNamedStress( bundleContext );
	stress.start();

	if ( !_testLatch.await( 15, TimeUnit.SECONDS ) )
	{

	log( LogService.LOG_DEBUG, "create latch: " + _factoryConfigCreateLatch.getCount() );
	log( LogService.LOG_DEBUG, "update latch: " + _factoryConfigUpdateLatch.getCount() );
	log( LogService.LOG_DEBUG, "delete latch: " + _factoryConfigDeleteLatch.getCount() );

	Assert.fail( "Test did not completed timely" );
	}
	}
	catch ( InterruptedException e )
	{
	Assert.fail( "Test interrupted" );
	}
	}


	/**
	* Setup the latches used throughout this test
	*/
	private void setupLatches()
	{
	_factoryConfigCreateLatch = new CountDownLatch( 1 );
	_factoryConfigUpdateLatch = new CountDownLatch( UPDATE_LOOP );
	_factoryConfigDeleteLatch = new CountDownLatch( 1 );
	}

	/**
	* This is our Factory class which will react up CM factory configuration objects.
	* Each time a factory configuration object is created, the _factoryConfigCreatedLatch is counted down.
	* Each time a factory configuration object is updated, the _factoryConfigUpdatedLatch is counted down.
	* Each time a factory configuration object is deleted, the _factoryConfigDeletedLatch is counted down.
	*/
	@Ignore
	class Factory implements ManagedServiceFactory
	{
	Set<String> _pids = new HashSet<String>();


	public synchronized void updated( String pid, Dictionary<String, ?> properties ) throws ConfigurationException
	{
	if ( _pids.add( pid ) )
	{
	// pid created
	_factoryConfigCreateLatch.countDown();
	log( LogService.LOG_DEBUG, "Config created; create latch= " + _factoryConfigCreateLatch.getCount() );
	}
	else
	{
	// pid updated
	try
	{
	Long number = ( Long ) properties.get( "number" );
	long currentNumber = _factoryConfigUpdateLatch.getCount();
	if ( number.longValue() != currentNumber )
	{
	throw new ConfigurationException( "number", "Expected number=" + currentNumber + ", actual="
	+ number );
	}
	_factoryConfigUpdateLatch.countDown();
	log( LogService.LOG_DEBUG, "Config updated; update latch= " + _factoryConfigUpdateLatch.getCount()
	+ " (number=" + number + ")" );
	}
	catch ( ClassCastException e )
	{
	throw new ConfigurationException( "number", e.getMessage(), e );
	}
	}
	}


	public void deleted( String pid )
	{
	// We need to remove this as the same PID can re-occur after deletion
	_pids.remove(pid);
	_factoryConfigDeleteLatch.countDown();
	log( LogService.LOG_DEBUG, "Config deleted; delete latch= " + _factoryConfigDeleteLatch.getCount() );
	}


	public String getName()
	{
	return "MyPID";
	}
	}

	/**
	* This class creates/update/delete some factory configuration instances, using a separate thread.
	*/
	@Ignore
	abstract class CreateStress extends Thread
	{
	BundleContext _bc;


	CreateStress( BundleContext bctx )
	{
	_bc = bctx;
	}


	public void run()
	{
	try
	{
	System.out.println( "Starting CM stress test ..." );
	ConfigurationAdmin cm = ( ConfigurationAdmin ) _tracker.waitForService( 2000 );
	setupLatches();
	Factory factory = new Factory();
	Hashtable<String, Object> serviceProps = new Hashtable<String, Object>();
	serviceProps.put( "service.pid", _FACTORYPID );
	_bc.registerService( ManagedServiceFactory.class.getName(), factory, serviceProps );

	for ( int l = 0; l < TEST_LOOP; l++ )
	{
	// Create factory configuration
	Hashtable<String, Object> props = new Hashtable<String, Object>();
	props.put( "foo", "bar" );
	obtainConfiguration(cm).update( props );

	// Check if our Factory has seen the factory configuration creation
	if ( !_factoryConfigCreateLatch.await( 10, TimeUnit.SECONDS ) )
	{
	throw new RuntimeException( "_factoryConfigCreateLatch did not reach zero timely" );
	}

	// Update factory configuration many times
	for ( int i = 0; i < UPDATE_LOOP; i++ )
	{
	props = new Hashtable<String, Object>();
	props.put( "foo", "bar" + i );
	props.put( "number", Long.valueOf( UPDATE_LOOP - i ) );
	obtainConfiguration(cm).update( props );
	}

	// Check if all configuration updates have been caught by our Factory
	if ( !_factoryConfigUpdateLatch.await( 10, TimeUnit.SECONDS ) )
	{
	throw new RuntimeException( "_factoryConfigUpdateLatch did not reach zero timely" );
	}

	// Remove factory configuration
	obtainConfiguration(cm).delete();

	// Check if our Factory has seen the configration removal
	if ( !_factoryConfigDeleteLatch.await( 10, TimeUnit.SECONDS ) )
	{
	throw new RuntimeException( "_factoryConfigDeleteLatch did not reach zero timely" );
	}

	// Reset latches
	setupLatches();
	}
	}
	catch ( Exception e )
	{
	e.printStackTrace( System.err );
	return;
	}
	_testLatch.countDown(); // Notify that our test is done
	}


	protected abstract org.osgi.service.cm.Configuration obtainConfiguration(ConfigurationAdmin cm)
	throws Exception;

	}

	@Ignore
	class CreateUpdateStress extends CreateStress {
	CreateUpdateStress(BundleContext bctx) {
	super(bctx);
	}

	protected org.osgi.service.cm.Configuration obtainConfiguration(ConfigurationAdmin cm)
	throws Exception {
	Configuration[] cfgs = cm.listConfigurations("(service.factoryPid=" + _FACTORYPID + ")" );

	org.osgi.service.cm.Configuration conf;
	if(cfgs == null) {
	conf = cm.createFactoryConfiguration( _FACTORYPID, null );
	} else if (cfgs.length == 1) {
	conf = cfgs[0];
	} else {
	throw new IllegalArgumentException("Only one configuration expected");
	}

	return conf;
	}
	}

	@Ignore
	class CreateUpdateNamedStress extends CreateStress {
	CreateUpdateNamedStress(BundleContext bctx) {
	super(bctx);
	}

	protected org.osgi.service.cm.Configuration obtainConfiguration(ConfigurationAdmin cm)
	throws IOException {
	org.osgi.service.cm.Configuration conf = cm.getFactoryConfiguration( _FACTORYPID, _FACTORYNAME, null );
	return conf;
	}
	}
	}