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 2014/06/03 17:51:25 UTC
svn commit: r1599630 -
/felix/sandbox/pderop/dependencymanager-prototype/org.apache.felix.dm/src/org/apache/felix/dm/impl/SerialExecutor.java
Author: pderop
Date: Tue Jun 3 15:51:25 2014
New Revision: 1599630
URL: http://svn.apache.org/r1599630
Log:
Code cleanup, and also made a refactoring for the support of a lock-free SerialExecutor (for performance reasons).
Modified:
felix/sandbox/pderop/dependencymanager-prototype/org.apache.felix.dm/src/org/apache/felix/dm/impl/SerialExecutor.java
Modified: felix/sandbox/pderop/dependencymanager-prototype/org.apache.felix.dm/src/org/apache/felix/dm/impl/SerialExecutor.java
URL: http://svn.apache.org/viewvc/felix/sandbox/pderop/dependencymanager-prototype/org.apache.felix.dm/src/org/apache/felix/dm/impl/SerialExecutor.java?rev=1599630&r1=1599629&r2=1599630&view=diff
==============================================================================
--- felix/sandbox/pderop/dependencymanager-prototype/org.apache.felix.dm/src/org/apache/felix/dm/impl/SerialExecutor.java (original)
+++ felix/sandbox/pderop/dependencymanager-prototype/org.apache.felix.dm/src/org/apache/felix/dm/impl/SerialExecutor.java Tue Jun 3 15:51:25 2014
@@ -1,8 +1,8 @@
package org.apache.felix.dm.impl;
-import java.util.LinkedList;
-import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicReference;
import org.osgi.service.log.LogService;
@@ -11,131 +11,126 @@ import org.osgi.service.log.LogService;
* them on one thread sequentially. It assumes more than one thread will
* try to execute the tasks and it will make an effort to pick the first
* task that comes along whilst making sure subsequent tasks return
- * without waiting.
+ * without waiting. <p>
+ *
+ * This class is lock free and ensures "safe object publication" between scheduling threads and
+ * actual executing thread: if one thread T1 schedules a task, but another thread T2 actually
+ * executes it, then all the objects from the T1 thread will be "safely published" to the executing T2 thread.
+ * Safe publication is ensured because we are using a ConcurrentLinkedQueue.
+ * (see [1], chapter 3.5.3 (Safe publication idioms).
+ *
+ * [1] http://www.cs.umd.edu/~pugh/java/memoryModel/jsr-133-faq.html#volatile
*
* @author <a href="mailto:dev@felix.apache.org">Felix Project Team</a>
*/
public class SerialExecutor implements Executor {
- private final List<Runnable> m_queue = new LinkedList<Runnable>();
- private Thread m_runningThread;
+ /**
+ * All tasks scheduled are stored there and only one thread may run them.
+ **/
+ protected final ConcurrentLinkedQueue<Runnable> m_tasks = new ConcurrentLinkedQueue<Runnable>();
+
+ /**
+ * Thread currently executing the task queue.
+ **/
+ protected final AtomicReference<Thread> m_runningThread = new AtomicReference<>();
+
+ /**
+ * Logger used when a task execution throws an exception
+ **/
private final Logger m_logger;
+ /**
+ * Makes a new SerialExecutor
+ * @param logger used when a task execution throws an exception. Can be null if no exception should be logger.
+ */
public SerialExecutor(Logger logger) {
m_logger = logger;
}
/**
- * Execute a task. This method is thread safe, so multiple threads can try to execute a task
- * but only the first will be executed, other threads will return immediately, and the
- * first thread will execute the tasks scheduled by the other threads.<p>
- * <p>
- * This method is reentrant: if the current thread is currently being executed by this executor, then any
- * subsequent tasks scheduled in this executor will executed immediately (inline execution).
+ * Enqueues a task for later execution. You must call {@link #execute()} in order
+ * to trigger the task execution, which may or may not be executed by
+ * your current thread.
*/
- public void execute(Runnable command) {
- execute(command, true);
+ public void schedule(Runnable task) {
+ m_tasks.add(task); // No need to synchronize, m_tasks is a concurrent linked queue.
}
-
+
/**
- * Execute a task. This method is thread safe, so multiple threads can try to execute a task
- * but only the first will be executed, other threads will return immediately, and the
- * first thread will execute the tasks scheduled by the other threads.
- *
- * @param command the task to execute (possibly to be delayed and executed by another single thread)
- * @param reentrant true means that if the executor is already being executed on this thread, then the
- * command is run immediately (inline execution). else, if reentrant == false, then the task is enqueued
- * even if the executor is currently being executed by the current thread.
+ * Executes any pending tasks, enqueued using the {@link SerialExecutor#schedule(Runnable)} method.
+ * This method is thread safe, so multiple threads can try to execute the pending
+ * tasks, but only the first will be used to actually do so. Other threads will return immediately.
*/
- public void execute(Runnable command, boolean reentrant) {
- Runnable next = null;
+ public void execute() {
Thread currentThread = Thread.currentThread();
- boolean isExecutingOnThisThread = false;
-
- synchronized (this) {
- if (reentrant && m_runningThread == currentThread) {
- // we are already being executed by this executor, so we'll execute the command directly (inline execution).
- isExecutingOnThisThread = true;
- }
- else {
- if (m_queue.isEmpty()) {
- // Nobody is currently using the executor, we'll execute the command, but we have to store it
- // in the queue, so another thread won't execute concurrently. Also, we mark the current executing
- // thread to our own current thread, in order to inline executions of reentrant scheduled tasks, if any.
- m_runningThread = currentThread;
- next = command;
- }
- m_queue.add(command);
- }
+ if (m_runningThread.compareAndSet(null, currentThread)) {
+ runTasks(currentThread);
}
+ }
- if (reentrant && isExecutingOnThisThread) {
- runTask(command);
+ /**
+ * Schedules a task for execution, and then attempts to execute it. This method is thread safe, so
+ * multiple threads can try to execute a task but only the first will be executed, other threads will
+ * return immediately, and the first thread will execute the tasks scheduled by the other threads.<p>
+ * <p>
+ * This method is reentrant: if the current thread is currently being executed by this executor, then
+ * the task passed to this method will be executed immediately, from the current invoking thread
+ * (inline execution).
+ */
+ public void execute(Runnable task) {
+ Thread currentThread = Thread.currentThread();
+ if (m_runningThread.get() == currentThread) {
+ runTask(task);
+ } else {
+ schedule(task);
+ execute();
}
- else {
- while (next != null) {
- runTask(next);
- synchronized (this) {
- m_queue.remove(0); // The first element is the one we have just executed
- next = m_queue.isEmpty() ? null : (Runnable) m_queue.get(0);
- if (next == null) {
- m_runningThread = null;
- }
+ }
+
+ /**
+ * Run all pending tasks
+ * @param currentRunninghread the current executing thread
+ */
+ private void runTasks(Thread currentRunninghread) {
+ do {
+ try {
+ // We do a memory barrier in order to ensure consistent per-thread memory visibility.
+ // Only one thread at a time is running this method, so there is no possible contention.
+ Runnable task;
+ ConcurrentLinkedQueue<Runnable> tasks = m_tasks;
+
+ while ((task = tasks.poll()) != null) {
+ runTask(task);
}
}
+ finally {
+ m_runningThread.set(null);
+ }
}
+ // We must test again if some tasks have been scheduled after our "while" loop above, but before the
+ // m_runningThread reference has been reset to null.
+ while (!m_tasks.isEmpty() && m_runningThread.compareAndSet(null, currentRunninghread));
}
- private void runTask(Runnable command) {
+ /**
+ * Run a given task.
+ * @param task the task to execute.
+ */
+ void runTask(Runnable command) {
try {
command.run();
}
catch (Throwable t) {
- if (m_logger != null) {
- m_logger.log(LogService.LOG_ERROR, "Error processing tasks", t);
- }
+ if (m_logger != null) {
+ m_logger.log(LogService.LOG_ERROR, "Error processing tasks", t);
+ } else {
+ t.printStackTrace();
+ }
}
}
- // TODO ASPECTS: Review. Added methods for separately scheduling and executing tasks
- // on the SerialExecutor. This is used in the ServiceTracker which must ensure
- // customizer callback methods are executed in the correct order.
-
-
- public void schedule(Runnable runnable) {
- synchronized (this) {
- m_queue.add(runnable);
- }
- }
-
- public void execute() {
- Runnable next = null;
- synchronized (this) {
- if (m_runningThread == null || m_runningThread == Thread.currentThread()) {
- // It's our turn
- if (!m_queue.isEmpty()) {
- next = m_queue.get(0);
- m_runningThread = Thread.currentThread();
- } else {
- return;
- }
- } else {
- return;
- }
- }
- while (next != null) {
- runTask(next);
- synchronized (this) {
- m_queue.remove(0); // The first element is the one we have just executed
- next = m_queue.isEmpty() ? null : (Runnable) m_queue.get(0);
- if (next == null) {
- m_runningThread = null;
- }
- }
- }
- }
-
- @Override
- public String toString() {
- return "[Executor: queue size: " + m_queue.size() + "]";
- }
+ @Override
+ public String toString() {
+ return "[Executor: queue size: " + m_tasks.size() + "]";
+ }
}
\ No newline at end of file