You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@felix.apache.org by pd...@apache.org on 2017/01/04 21:32:42 UTC
svn commit: r1777378 - in /felix/trunk/dependencymanager:
org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/
org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/util/
org.apache.felix.dependencymanager/src/or...
Author: pderop
Date: Wed Jan 4 21:32:42 2017
New Revision: 1777378
URL: http://svn.apache.org/viewvc?rev=1777378&view=rev
Log:
FELIX-5471: added javadoc for the schedule method in ComponentImpl class (and renamed "synchronous" flag to "bypassThreadPoolIfPossible").
Added ServiceRaceWithOrderedUnbindTest and ServiceRaceParallelWithOrderedUnbindTest, which verify the following: if M depends on X, and if
X is removed, then M.unbind(X) is called before X.stop() is called (this can only happen if X is removed when all services are fully started.
Added:
felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceParallelWithOrderedUnbindTest.java
felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceWithOrderedUnbindTest.java
Modified:
felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/util/TestBase.java
felix/trunk/dependencymanager/org.apache.felix.dependencymanager/src/org/apache/felix/dm/impl/ComponentImpl.java
Added: felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceParallelWithOrderedUnbindTest.java
URL: http://svn.apache.org/viewvc/felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceParallelWithOrderedUnbindTest.java?rev=1777378&view=auto
==============================================================================
--- felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceParallelWithOrderedUnbindTest.java (added)
+++ felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceParallelWithOrderedUnbindTest.java Wed Jan 4 21:32:42 2017
@@ -0,0 +1,38 @@
+/*
+ * 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.dm.itest.api;
+
+/**
+ * This test class simulates a client having many dependencies being registered concurrently using DM concurrent mode.
+ * (that is: a ComponentExecutorFactory is registered in the osgi registry and the configured threadpool is then used
+ * to schedule component creations).
+ * Once the services are created, injected, and fully started, then they are unregistered from a single thread (like it is the case when the osgi
+ * framework is stopped where bundle are stopped synchronously).
+ * So, when unbind methods are called, we verify that unbound services are still started at the time unbind callbacks are invoked.
+ *
+ * NOTICE: when using DM and a ComponentExecutorFactory, component removal is done synchronously if possible.
+ * So if you remove components after all components have been fully started, then the component removal are done synchronously.
+ *
+ * @author <a href="mailto:dev@felix.apache.org">Felix Project Team</a>
+ */
+public class ServiceRaceParallelWithOrderedUnbindTest extends ServiceRaceWithOrderedUnbindTest {
+ public ServiceRaceParallelWithOrderedUnbindTest() {
+ setParallel(); // Configure DM to use a threadpool
+ }
+}
Added: felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceWithOrderedUnbindTest.java
URL: http://svn.apache.org/viewvc/felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceWithOrderedUnbindTest.java?rev=1777378&view=auto
==============================================================================
--- felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceWithOrderedUnbindTest.java (added)
+++ felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/api/ServiceRaceWithOrderedUnbindTest.java Wed Jan 4 21:32:42 2017
@@ -0,0 +1,255 @@
+/*
+ * 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.dm.itest.api;
+
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.felix.dm.Component;
+import org.apache.felix.dm.ComponentStateListener;
+import org.apache.felix.dm.ServiceDependency;
+import org.apache.felix.dm.itest.util.Ensure;
+import org.apache.felix.dm.itest.util.TestBase;
+import org.junit.Assert;
+
+
+/**
+ * This test class simulates a client having many dependencies being registered concurrently.
+ * (threads are created manually, and we are not using a ComponentExecutorFactory).
+ * The services are then unregistered from a single thread (like it is the case when the osgi
+ * framework is stopped where bunldes are stopped synchronously).
+ * So, when unbind methods are called, we verify that unbound services are still started at
+ * the time unbind callbacks are invoked.
+ *
+ * @author <a href="mailto:dev@felix.apache.org">Felix Project Team</a>
+ */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class ServiceRaceWithOrderedUnbindTest extends TestBase {
+ final static int STEP_WAIT = 5000;
+ final static int DEPENDENCIES = 10;
+ final static int LOOPS = 3000;
+ final Ensure m_done = new Ensure(true);
+
+ // Timestamp used to log the time consumed to execute 100 tests.
+ long m_timeStamp;
+
+ public interface Dep {
+ boolean isStarted();
+ }
+
+ public class DepImpl implements Dep {
+ volatile boolean m_started;
+
+ void start() {
+ m_started = true;
+ }
+
+ void stop() {
+ m_started = false;
+ }
+
+ public boolean isStarted() {
+ return m_started;
+ }
+ }
+
+ /**
+ * Creates many service dependencies, and activate/deactivate them concurrently.
+ */
+ public void testCreatesComponentsConcurrently() {
+ m_dm.add(m_dm.createComponent()
+ .setImplementation(this)
+ .setCallbacks(null, "start", null, null));
+ m_done.waitForStep(1, 60000);
+ m_dm.clear();
+ Assert.assertFalse(super.errorsLogged());
+ }
+
+ void start() {
+ new Thread(this::doStart).start();
+ }
+
+ void doStart() {
+ info("Starting createParallelComponentRegistgrationUnregistration test");
+ initThreadPool(); // only if setParallel() has not been called (only if a parallel DM is not used).
+
+ try {
+ m_timeStamp = System.currentTimeMillis();
+ for (int loop = 0; loop < LOOPS; loop++) {
+ doTest(loop);
+ }
+ }
+ catch (Throwable t) {
+ error("got unexpected exception", t);
+ }
+ finally {
+ shutdownThreadPool();
+ m_done.step(1);
+ }
+ }
+
+ private void initThreadPool() {
+ if (! m_parallel) {
+ // We are not using a parallel DM, so we create a custom threadpool in order to add components concurrently.
+ int cores = Math.max(16, Runtime.getRuntime().availableProcessors());
+ info("using " + cores + " cores.");
+ m_threadPool = new ForkJoinPool(Math.max(cores, DEPENDENCIES + 3 /* start/stop/configure */));
+ }
+ }
+
+ void shutdownThreadPool() {
+ if (! m_parallel && m_threadPool != null) {
+ m_threadPool.shutdown();
+ try {
+ m_threadPool.awaitTermination(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+
+ void doTest(int loop) throws Throwable {
+ debug("loop#%d -------------------------", loop);
+
+ final Ensure step = new Ensure(false);
+
+ // Create one client component, which depends on many service dependencies
+ final Component client = m_dm.createComponent();
+ final Client clientImpl = new Client(step);
+ client.setImplementation(clientImpl);
+
+ // Before creating the client, register a component listener to check
+ // the client is really started or deactivated.
+ ComponentStateListener clientListener = (c, s) -> {
+ switch(s) {
+ case TRACKING_OPTIONAL:
+ step.step(1);
+ break;
+ case INACTIVE:
+ step.step(2);
+ break;
+ default:
+ break;
+ }
+ };
+ client.add(clientListener);
+
+ // Create client service dependencies
+ final ServiceDependency[] dependencies = new ServiceDependency[DEPENDENCIES];
+ for (int i = 0; i < DEPENDENCIES; i++) {
+ final String filter = "(id=loop" + loop + "." + i + ")";
+ dependencies[i] = m_dm.createServiceDependency().setService(Dep.class, filter)
+ .setRequired(true)
+ .setCallbacks("add", "remove");
+ client.add(dependencies[i]);
+ }
+
+ // Activate the client service dependencies concurrently.
+ List<Component> deps = new ArrayList();
+ for (int i = 0; i < DEPENDENCIES; i++) {
+ Hashtable h = new Hashtable();
+ h.put("id", "loop" + loop + "." + i);
+ final Component s = m_dm.createComponent()
+ .setInterface(Dep.class.getName(), h)
+ .setImplementation(new DepImpl());
+ deps.add(s);
+ schedule(() -> m_dm.add(s));
+ }
+
+ // Start the client (concurrently)
+ schedule(() -> m_dm.add(client));
+
+ // Ensure that client has been started.
+ step.waitForStep(1, STEP_WAIT); // client has entered in TRACKING_OPTIONAL state
+ Assert.assertEquals(DEPENDENCIES, clientImpl.getDependencies());
+
+ // Make sure threadpool is quiescent, then deactivate all components.
+ if (! m_threadPool.awaitQuiescence(5000, TimeUnit.MILLISECONDS)) {
+ throw new RuntimeException("Could not start components timely.");
+ }
+
+ // Stop all dependencies, and client
+ schedule(() -> {
+ for (Component dep : deps) {
+ final Component dependency = dep;
+ m_dm.remove(dependency);
+ }
+ m_dm.remove(client);
+ });
+
+ // Ensure that client has been stopped, then destroyed, then unbound from all dependencies
+ step.waitForStep(2, STEP_WAIT); // Client entered in INACTIVE state
+ step.ensure();
+ Assert.assertEquals(0, clientImpl.getDependencies());
+
+ // Make sure threadpool is quiescent before doing next iteration.
+ if (! m_threadPool.awaitQuiescence(5000, TimeUnit.MILLISECONDS)) {
+ throw new RuntimeException("Could not start components timely.");
+ }
+
+ if (super.errorsLogged()) {
+ throw new IllegalStateException("Race test interrupted (some error occured, see previous logs)");
+ }
+
+ debug("finished one test loop");
+ if ((loop + 1) % 100 == 0) {
+ long duration = System.currentTimeMillis() - m_timeStamp;
+ warn("Performed 100 tests (total=%d) in %d ms.", (loop + 1), duration);
+ m_timeStamp = System.currentTimeMillis();
+ }
+ }
+
+ private void schedule(Runnable task) {
+ if (! m_parallel) {
+ // not using parallel DM, so use our custom threadpool.
+ m_threadPool.execute(task);
+ } else {
+ task.run();
+ }
+ }
+
+ public class Client {
+ final Ensure m_step;
+ volatile int m_dependencies;
+
+ public Client(Ensure step) {
+ m_step = step;
+ }
+
+ void add(Dep d) {
+ Assert.assertNotNull(d);
+ m_dependencies ++;
+ }
+
+ void remove(Dep d) {
+ Assert.assertNotNull(d);
+ if (! d.isStarted()) {
+ Thread.dumpStack();
+ }
+ Assert.assertTrue(d.isStarted());
+ m_dependencies --;
+ }
+
+ int getDependencies() {
+ return m_dependencies;
+ }
+ }
+}
Modified: felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/util/TestBase.java
URL: http://svn.apache.org/viewvc/felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/util/TestBase.java?rev=1777378&r1=1777377&r2=1777378&view=diff
==============================================================================
--- felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/util/TestBase.java (original)
+++ felix/trunk/dependencymanager/org.apache.felix.dependencymanager.itest/src/org/apache/felix/dm/itest/util/TestBase.java Wed Jan 4 21:32:42 2017
@@ -24,6 +24,7 @@ import java.util.Hashtable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
@@ -53,7 +54,7 @@ public abstract class TestBase extends T
protected final static int LOG_LEVEL = LogService.LOG_WARNING;
// optional thread pool used by parallel dependency managers
- private volatile ExecutorService m_threadPool;
+ protected volatile ForkJoinPool m_threadPool;
// flag used to check if the threadpool must be used for a given test.
protected volatile boolean m_parallel;
@@ -90,7 +91,7 @@ public abstract class TestBase extends T
m_dm = new DependencyManager(context);
if (m_parallel) {
warn("Using threadpool ...");
- m_threadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
+ m_threadPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
m_componentExecutorFactoryReg = context.registerService(ComponentExecutorFactory.class.getName(),
new ComponentExecutorFactory() {
@Override
Modified: felix/trunk/dependencymanager/org.apache.felix.dependencymanager/src/org/apache/felix/dm/impl/ComponentImpl.java
URL: http://svn.apache.org/viewvc/felix/trunk/dependencymanager/org.apache.felix.dependencymanager/src/org/apache/felix/dm/impl/ComponentImpl.java?rev=1777378&r1=1777377&r2=1777378&view=diff
==============================================================================
--- felix/trunk/dependencymanager/org.apache.felix.dependencymanager/src/org/apache/felix/dm/impl/ComponentImpl.java (original)
+++ felix/trunk/dependencymanager/org.apache.felix.dependencymanager/src/org/apache/felix/dm/impl/ComponentImpl.java Wed Jan 4 21:32:42 2017
@@ -421,7 +421,8 @@ public class ComponentImpl implements Co
@Override
public void stop() {
if (m_active.compareAndSet(true, false)) {
- schedule(true /* try execute synchronously if using a tpool */, () -> {
+ // try to be synchronous, even if a threadpool is used (best effort).
+ schedule(true /* bypass threadpool if possible */, () -> {
m_isStarted = false;
handleChange();
});
@@ -457,8 +458,8 @@ public class ComponentImpl implements Co
// This is just a best effort, and the removed event will be handled asynchronosly if our
// queue is currently being run by another thread, or by the threadpool.
- boolean synchronously = (type == EventType.REMOVED);
- schedule(synchronously, () -> {
+ boolean bypassThreadPoolIfPossible = (type == EventType.REMOVED);
+ schedule(bypassThreadPoolIfPossible, () -> {
try {
switch (type) {
case ADDED:
@@ -1677,24 +1678,34 @@ public class ComponentImpl implements Co
}
/**
- * Executes a task using our queue. The task is executed synchronously in case the queue is
- * not being run by another thread, or by the threadpool.
+ * Executes a task using our component queue. The queue associated to this component is by default a "SerialExecutor", or
+ * a "DispatchExecutor" if a threadpool is used (that is: if a ComponentExecutorFactory has returned an executor this our
+ * component).
*
- * @param trySynchronous try to execute the task synchronously (best effort).
+ * A SerialExecutor always try to execute the task synchronously, except if another master thread is currently
+ * running the SerialExecutor queue.
+ * A DispatchExecutor always schedule the task asynchronously in a threadpool, but can optionally try to execute the task synchronously
+ * (the threadpool is bypassed only if the queue is not currently being run by the threadpool).
+ *
+ * The "bypassThreadPoolIfPossible" argument is only used in case there is a threadpool configured. If no threadpool is used,
+ * this argument is ignored and the task is run synchronously if the serial queue is not concurrently run from another
+ * master threads.
+ *
+ * @param bypassThreadPoolIfPossible This argument is only used if a threadpool is configured for this component. if true,
+ * it means that if a threadpool is configured, then we attempt to run the task synchronously and not using the threadpool,
+ * but if the queue is currently run from the threadpool, then the task is scheduled in it.
+ * false means we always use the threadpool and the task is executed asynchronously.
* @param task the task to execute.
*/
- private void schedule(boolean trySynchronous, Runnable task) {
+ private void schedule(boolean bypassThreadPoolIfPossible, Runnable task) {
Executor exec = getExecutor();
- if (trySynchronous) {
- if (exec instanceof DispatchExecutor) {
- // Try to execute the task from the current thread if the threadpool is not currently running our queue.
- ((DispatchExecutor) exec).execute(task, false);
- return;
- }
+ if (exec instanceof DispatchExecutor) {
+ // We are using a threadpool. If byPassThreadPoolIsPossible is true, it means we can try to run the task synchronously from the current
+ // thread. But if our queue is currently run from the threadpool, then the task will be run from it (asynchronously).
+ ((DispatchExecutor) exec).execute(task, ! bypassThreadPoolIfPossible);
+ } else {
+ // The queue is a serial queue: the task will be run synchronously, except if another master thread is currently running the queue.
+ exec.execute(task);
}
- // If we are using a serial queue (no threadpool), then the queue executes the task synchronously
- // if no other master thread is running the queue.
- // If the are using a threadpool, then the task is always executed asynchronously, in the threadpool.
- exec.execute(task);
}
}