1 files changed, 238 insertions, 228 deletions

diff --git a/src/main/java/org/gnunet/util/Scheduler.java b/src/main/java/org/gnunet/util/Scheduler.java
index f5eee8c..1f8d9dc 100644
--- a/src/main/java/org/gnunet/util/Scheduler.java
+++ b/src/main/java/org/gnunet/util/Scheduler.java
@@ -20,6 +20,7 @@
 
 package org.gnunet.util;
 
+import com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,10 +29,13 @@ import java.nio.ByteBuffer;
 import java.nio.channels.*;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.*;
-import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Schedule computations using continuation passing style.
+ * <p/>
+ * By design, it is not safe to scheduler tasks from different threads.
+ * When threads can't be avoided, they should communicate with the scheduler by
+ * a pipe.
  *
  * @author Florian Dold
  */
@@ -39,12 +43,10 @@ public class Scheduler {
     private static final Logger logger = LoggerFactory
             .getLogger(Scheduler.class);
 
-    private static ReentrantLock lock = new ReentrantLock(true);
-
     /**
      * Task that we are currently executing, or null if no task is currently running.
      */
-    private static TaskConfiguration activeTask = null;
+    private static TaskIdentifier activeTask = null;
 
     /**
      * Number of tasks in the ready lists, that is, number of tasks that is ready to run
@@ -69,21 +71,14 @@ public class Scheduler {
      * For every priority, there is a list of tasks that is definitely ready to run.
      */
     @SuppressWarnings("unchecked")
-    final private static LinkedList<TaskConfiguration>[] readyLists = new LinkedList[Priority.numberOfPriorities];
+    final private static LinkedList<TaskIdentifier>[] readyLists = new LinkedList[Priority.numberOfPriorities];
 
     static {
         for (int i = 0; i < Priority.numberOfPriorities; ++i) {
-            readyLists[i] = new LinkedList<TaskConfiguration>();
+            readyLists[i] = new LinkedList<TaskIdentifier>();
         }
     }
 
-    private static final int EVENT_READ = 0, EVENT_WRITE = 1, EVENT_ACCEPT = 2, EVENT_CONNECT = 3;
-    private static final int[] eventToInterestOp = new int[]{SelectionKey.OP_READ, SelectionKey.OP_WRITE,
-            SelectionKey.OP_ACCEPT, SelectionKey.OP_CONNECT};
-    private static final Reason[] eventToReason = new Reason[]{Reason.READ_READY, Reason.WRITE_READY,
-            Reason.ACCEPT_READY, Reason.CONNECT_READY};
-
-
     /**
      * Selector, used to check file descriptors for readiness.
      */
@@ -102,37 +97,20 @@ public class Scheduler {
     /**
      * true iff the scheduler is currently running.
      */
-    private static boolean scheduler_running = false;
-
+    private static boolean schedulerRunning = false;
 
     /**
      * Pending tasks are waiting for an event. Each pending task has a (possibly infinitely long)
      * deadline after which the task is executed regardless of the prerequisites.
      */
-    final private static Queue<TaskConfiguration> pending = new PriorityQueue<TaskConfiguration>(5, new Comparator
-            <TaskConfiguration>() {
+    final private static Queue<TaskIdentifier> pending = new PriorityQueue<TaskIdentifier>(5, new Comparator
+            <TaskIdentifier>() {
         @Override
-        public int compare(TaskConfiguration a, TaskConfiguration b) {
+        public int compare(TaskIdentifier a, TaskIdentifier b) {
             return a.deadline.compareTo(b.deadline);
         }
     });
 
-
-    /**
-     * Reset the scheduler forcefully.
-     * Intended to be used internally in the Scheduler, as well as in test teardown.
-     */
-    public static void forceReset() {
-        scheduler_running = false;
-        readyCount = 0;
-        activeTask = null;
-        for (int i = 0; i < Priority.numberOfPriorities; ++i) {
-            readyLists[i] = new LinkedList<TaskConfiguration>();
-        }
-        pending.clear();
-    }
-
-
     /**
      * Reasons for executing a task.
      */
@@ -147,94 +125,133 @@ public class Scheduler {
         /**
          * The reason this task has been called by the scheduler.
          */
-        public Set<Reason> reasons = EnumSet.noneOf(Reason.class);
+        public EnumSet<Reason> reasons = EnumSet.noneOf(Reason.class);
     }
 
     /**
-     * A task is the basic unit of work that is managed by the scheduler.
+     * Which operations is a task identifier interested in?
      */
-    public static interface Task {
-        public void run(RunContext ctx);
+    private static class TaskInterestOps {
+        TaskIdentifier tid;
+        int interestOps;
     }
 
     /**
-     * A TaskConfiguration represents a Task that will execute or has already been executed.
+     * Manage subscriptions for selection events on channels.
      */
-    public static class TaskConfiguration implements Cancelable {
-        private final Task task;
-        private RunContext ctx = new RunContext();
-        private boolean lifeness = true;
-        private Priority priority;
-        private final AbsoluteTime deadline;
-
-        private ArrayList<SelectableChannel> eventChannels = null;
-        private ArrayList<Integer> eventTypes = null;
-
-        private boolean hasRun = false;
-        private boolean isCanceled = false;
-
-        /**
-         * Create a TaskIdentifier.
-         *
-         * @param delay when will the task be run?
-         *              may be null to indicate that this task may not be run
-         *              (but only queued directly)
-         * @param task task to run with this TaskIdentifier
-         */
-        TaskConfiguration(RelativeTime delay, Task task) {
-            this.task = task;
-            if (delay == null)
-                this.deadline = null;
-            else
-                this.deadline = delay.toAbsolute();
+    private static class Subscriptions {
+        private static class ChannelInterest {
+            SelectableChannel channel;
+            int interestOps;
         }
 
-        /**
-         * Register the task configuration to run once the given event completed.
-         *
-         * @param channel channel that we wait for an event on
-         * @param eventType the event we wait on
-         */
-        private void addChannelEvent(SelectableChannel channel, int eventType) {
-            if (channel == null) {
-                throw new AssertionError("channel must be non-null");
+        List<ChannelInterest> channelInterests = Lists.newLinkedList();
+
+        void add(SelectableChannel channel, int interestOps) {
+            boolean found = false;
+            for (ChannelInterest ci : channelInterests) {
+                if (ci.channel == channel) {
+                    ci.interestOps |= interestOps;
+                    if ((ci.interestOps | SelectionKey.OP_CONNECT | SelectionKey.OP_READ) != 0) {
+                        throw new AssertionError("OP_CONNECT and OP_READ are incompatible in java");
+                    }
+                    found = true;
+                    break;
+                }
             }
-            if (eventChannels == null) {
-                eventChannels = new ArrayList<SelectableChannel>();
-                eventTypes = new ArrayList<Integer>();
+            if (!found) {
+                ChannelInterest ci = new ChannelInterest();
+                ci.channel = channel;
+                ci.interestOps = interestOps;
+                channelInterests.add(ci);
             }
-            eventChannels.add(channel);
-            eventTypes.add(eventType);
+        }
 
-            int interestOp = eventToInterestOp[eventType];
+        void apply(TaskIdentifier tid) {
+            for (ChannelInterest ci : channelInterests) {
+                SelectionKey key = ci.channel.keyFor(selector);
+                if (key == null || !key.isValid()) {
+                    try {
+                        key = ci.channel.register(selector, ci.interestOps);
+                        key.attach(new LinkedList());
+                    } catch (ClosedChannelException e) {
+                        throw new IOError(e);
+                    }
+                } else {
+                    key.interestOps(key.interestOps() | ci.interestOps);
+                }
+                @SuppressWarnings("unchecked")
+                LinkedList<TaskInterestOps> opl = (LinkedList<TaskInterestOps>) key.attachment();
+                TaskInterestOps tio = new TaskInterestOps();
+                tio.tid = tid;
+                tio.interestOps = ci.interestOps;
+                opl.add(tio);
+            }
+        }
 
-            SelectionKey key = channel.keyFor(selector);
-            if (key == null || !key.isValid()) {
-                try {
-                    // tread safety ... avoid deadlock
-                    selector.wakeup();
-                    key = channel.register(selector, interestOp, new TaskConfiguration[4]);
-                } catch (ClosedChannelException e) {
-                    throw new IOError(e);
+        void stop(TaskIdentifier tid) {
+            for (ChannelInterest ci : channelInterests) {
+                SelectionKey key = ci.channel.keyFor(selector);
+                if (key == null || !key.isValid()) {
+                    logger.warn("missing selection key");
+                    return;
                 }
-            } else {
-                if ((key.interestOps() & interestOp) != 0) {
-                    throw new AssertionError("interest op registered twice");
+                @SuppressWarnings("unchecked")
+                LinkedList<TaskInterestOps> interestList = (LinkedList<TaskInterestOps>) key.attachment();
+                Iterator<TaskInterestOps> it = interestList.iterator();
+                int remainingInterestOps = 0;
+                while (it.hasNext()) {
+                    TaskInterestOps ops = it.next();
+                    if (ops.tid == tid) {
+                        it.remove();
+                    } else {
+                        remainingInterestOps |= ops.interestOps;
+                    }
                 }
-                key.interestOps(key.interestOps() | interestOp);
+                key.interestOps(remainingInterestOps);
             }
+        }
+    }
 
-            TaskConfiguration[] subscribers = (TaskConfiguration[]) key.attachment();
-            if (subscribers[eventType] != null) {
-                throw new AssertionError("subscriber registered twice");
-            }
-            subscribers[eventType] = this;
+    /**
+     * A task is the basic unit of work that is managed by the scheduler.
+     */
+    public static interface Task {
+        public void run(RunContext ctx);
+    }
 
-            if (subscribers[EVENT_CONNECT] != null && subscribers[EVENT_READ] != null) {
-                throw new AssertionError("OP_CONNECT and OP_READ are incompatible in java");
-            }
+    /**
+     * Representation of a task that has been scheduled, and can be canceled
+     * until the task has run.
+     */
+    public static class TaskIdentifier implements Cancelable {
+        private boolean hasRun = false;
+        private boolean isCanceled = false;
+        private final Task task;
+        private final RunContext ctx = new RunContext();
+        private final boolean lifeness;
+        private final Priority priority;
+        private final AbsoluteTime deadline;
+        private final Subscriptions subscriptions;
+
+        public TaskIdentifier(Task task, EnumSet<Reason> reasons) {
+            this.ctx.reasons = reasons;
+            this.task = task;
+            lifeness = true;
+            priority = Priority.DEFAULT;
+            deadline = null;
+            subscriptions = null;
+        }
+
+        public TaskIdentifier(TaskConfiguration tc) {
+            this.task = tc.task;
+            this.subscriptions = tc.subscriptions;
+            this.deadline = tc.deadline;
+            this.priority = tc.priority;
+            this.lifeness = tc.lifeness;
         }
 
+
         private void run() {
             if (hasRun) {
                 throw new AssertionError("same task ran twice");
@@ -242,14 +259,18 @@ public class Scheduler {
             if (isCanceled) {
                 return;
             }
-            TaskConfiguration old = activeTask;
+            TaskIdentifier old = activeTask;
             activeTask = this;
             task.run(ctx);
             hasRun = true;
             activeTask = old;
         }
 
+        @Override
         public void cancel() {
+            if (hasRun) {
+                throw new AssertionError("can't cancel task that already ran");
+            }
             if (isCanceled) {
                 throw new AssertionError("task canceled twice");
             }
@@ -257,55 +278,59 @@ public class Scheduler {
             pending.remove(this);
         }
 
-        public Cancelable schedule() {
-            lock.lock();
-            if (this.deadline == null)
-                throw new AssertionError("a task without deadline may not be scheduled");
-            if (priority == null) {
-                if (activeTask != null) {
-                    priority = activeTask.priority;
-                } else {
-                    priority = Priority.DEFAULT;
-                }
+        public void deregister() {
+            if (subscriptions != null) {
+                subscriptions.stop(this);
             }
-            pending.add(this);
-            lock.unlock();
-            return this;
         }
+    }
 
-        private void deregister() {
-            if (eventChannels == null) {
-                return;
-            }
-            lock.lock();
-            for (int i = 0; i < eventChannels.size(); ++i) {
-                SelectionKey key = eventChannels.get(i).keyFor(selector);
-                TaskConfiguration[] subscribers = (TaskConfiguration[]) key.attachment();
-                int interestOp = eventToInterestOp[eventTypes.get(i)];
-                if (subscribers[eventTypes.get(i)] == null || (key.interestOps() | interestOp) == 0) {
-                    throw new AssertionError("deregistering event that has not been registered");
-                }
-                subscribers[eventTypes.get(i)] = null;
-                key.interestOps(key.interestOps() & (~interestOp));
-            }
+    /**
+     * A TaskConfiguration contains all information to schedule a task.
+     */
+    public static class TaskConfiguration {
+        private final Task task;
+        private boolean lifeness = true;
+        private Priority priority;
+        private final AbsoluteTime deadline;
 
-            lock.unlock();
-        }
+        private Subscriptions subscriptions;
 
-        public void selectRead(SelectableChannel channel) {
-            addChannelEvent(channel, EVENT_READ);
+        /**
+         * Create a TaskIdentifier.
+         *
+         * @param delay when will the task be run?
+         *              may be null to indicate that this task may not be run
+         *              (but only queued directly)
+         * @param task  task to run with this TaskIdentifier
+         */
+        TaskConfiguration(RelativeTime delay, Task task) {
+            if (delay == null)
+                throw new AssertionError("task delay may not be 'null'");
+            this.task = task;
+            this.deadline = delay.toAbsolute();
         }
 
-        public void selectWrite(SelectableChannel channel) {
-            addChannelEvent(channel, EVENT_WRITE);
-        }
+        public TaskIdentifier schedule() {
 
-        public void selectConnect(SelectableChannel channel) {
-            addChannelEvent(channel, EVENT_CONNECT);
+            if (priority == null) {
+                if (activeTask != null) {
+                    priority = activeTask.priority;
+                } else {
+                    priority = Priority.DEFAULT;
+                }
+            }
+            TaskIdentifier tid = new TaskIdentifier(this);
+            if (subscriptions != null)
+                subscriptions.apply(tid);
+            pending.add(tid);
+            return tid;
         }
 
-        public void selectAccept(SelectableChannel channel) {
-            addChannelEvent(channel, EVENT_ACCEPT);
+        public void addSelectEvent(SelectableChannel channel, int event) {
+            if (subscriptions == null)
+                subscriptions = new Subscriptions();
+            subscriptions.add(channel, event);
         }
     }
 
@@ -314,14 +339,7 @@ public class Scheduler {
      * the same priority.
      */
     public static void addContinuation(Task task, EnumSet<Reason> reasons) {
-        lock.lock();
-        TaskConfiguration t = new TaskConfiguration(null, task);
-        t.ctx.reasons = reasons;
-        t.priority = Priority.DEFAULT;
-        queueReady(t);
-        logger.debug("about to wake up");
-        lock.unlock();
-        selector.wakeup();
+        queueReady(new TaskIdentifier(task, reasons));
     }
 
     /**
@@ -343,26 +361,41 @@ public class Scheduler {
      * @param task  the task to run after delay
      * @return the TaskIdentifier, can be used to cancel the task until it has been executed.
      */
-    public static TaskConfiguration addDelayed(RelativeTime delay, Task task) {
+    public static TaskIdentifier addDelayed(RelativeTime delay, Task task) {
         TaskConfiguration tid = new TaskConfiguration(delay, task);
-        tid.schedule();
-        return tid;
+        return tid.schedule();
     }
 
-    public static TaskConfiguration addRead(RelativeTime timeout,
-                                            SelectableChannel chan, Task task) {
+    /**
+     * Add a task to run after the specified delay, or after the given channel
+     * is ready to read, whichever occurs first.
+     *
+     * @param timeout time to wait until running the task
+     * @param chan    chennel of interest
+     * @param task    task to run
+     * @return task identifier
+     */
+    public static TaskIdentifier addRead(RelativeTime timeout,
+                                         SelectableChannel chan, Task task) {
         TaskConfiguration tid = new TaskConfiguration(timeout, task);
-        tid.addChannelEvent(chan, EVENT_READ);
-        tid.schedule();
-        return tid;
+        tid.addSelectEvent(chan, SelectionKey.OP_READ);
+        return tid.schedule();
     }
 
-    public static TaskConfiguration addWrite(RelativeTime timeout,
-                                             SelectableChannel chan, Task task) {
+    /**
+     * Add a task to run after the specified delay, or after the given channel
+     * is ready to write, whichever occurs first.
+     *
+     * @param timeout to wait until running the task
+     * @param chan    channel of interest
+     * @param task    task to run
+     * @return task identifier
+     */
+    public static TaskIdentifier addWrite(RelativeTime timeout,
+                                          SelectableChannel chan, Task task) {
         TaskConfiguration tid = new TaskConfiguration(timeout, task);
-        tid.addChannelEvent(chan, EVENT_WRITE);
-        tid.schedule();
-        return tid;
+        tid.addSelectEvent(chan, SelectionKey.OP_WRITE);
+        return tid.schedule();
     }
 
     /**
@@ -372,14 +405,11 @@ public class Scheduler {
      * @return true to continue the main loop, false to exit
      */
     private static boolean checkLiveness() {
-        lock.lock();
         if (readyCount > 0) {
-            lock.unlock();
             return true;
         }
-        for (TaskConfiguration t : pending) {
+        for (TaskIdentifier t : pending) {
             if (t.lifeness) {
-                lock.unlock();
                 return true;
             }
         }
@@ -387,30 +417,23 @@ public class Scheduler {
         if (!pending.isEmpty()) {
             logger.debug("tasks pending but not alive -- disconnect");
             shutdown();
-            lock.unlock();
             return true;
         }
-
-        lock.unlock();
         return false;
     }
 
-
     /**
      * Queue a Task for execution.
      *
      * @param tid TaskIdentifier of the ready task
      */
-    private static void queueReady(TaskConfiguration tid) {
-        lock.lock();
+    private static void queueReady(TaskIdentifier tid) {
         int idx = tid.priority.ordinal();
         readyLists[idx].add(tid);
         readyCount++;
         pending.remove(tid);
-        lock.unlock();
     }
 
-
     /**
      * Queue all tasks with expired timeout.
      *
@@ -418,11 +441,10 @@ public class Scheduler {
      */
     private static RelativeTime handleTimeouts() {
         RelativeTime timeout = RelativeTime.FOREVER;
-        lock.lock();
 
         // check if any timeouts occurred
         while (true) {
-            TaskConfiguration t = pending.peek();
+            TaskIdentifier t = pending.peek();
             if (t == null) {
                 break;
             }
@@ -436,25 +458,19 @@ public class Scheduler {
                 break;
             }
         }
-        lock.unlock();
         return timeout;
     }
 
-    /**
-     * If there is a subscribing task for the given event type, add it to the set of executable tasks.
-     *
-     * @param executableTasks set of executable tasks
-     * @param subscribers subscriber set, one subscriber for each event type
-     * @param eventType event type we are interested in
-     */
-    private static void addSubscriberTask(Collection<TaskConfiguration> executableTasks,
-                                          TaskConfiguration[] subscribers, int eventType) {
-        TaskConfiguration tc = subscribers[eventType];
-        if (tc == null) {
-            return;
-        }
-        executableTasks.add(tc);
-        tc.ctx.reasons.add(eventToReason[eventType]);
+
+    private static void addReasonsFromInterestOp(EnumSet<Reason> reasons, int interestOps) {
+        if ((interestOps & SelectionKey.OP_READ) != 0)
+            reasons.add(Reason.READ_READY);
+        if ((interestOps & SelectionKey.OP_WRITE) != 0)
+            reasons.add(Reason.WRITE_READY);
+        if ((interestOps & SelectionKey.OP_CONNECT) != 0)
+            reasons.add(Reason.CONNECT_READY);
+        if ((interestOps & SelectionKey.OP_ACCEPT) != 0)
+            reasons.add(Reason.ACCEPT_READY);
     }
 
     /**
@@ -463,10 +479,8 @@ public class Scheduler {
      * @param timeout timeout for select
      */
     private static void handleSelect(RelativeTime timeout) {
-        if (!lock.isHeldByCurrentThread())
-            throw new AssertionError();
+        // gnunet-java uses microseconds, but the select api uses milliseconds
         long timeout_ms = timeout.getMicroseconds() / 1000;
-        lock.unlock();
         try {
             // selector.select(0) would block indefinitely (counter-intuitive, java's fault)
             if (timeout_ms == 0) {
@@ -480,39 +494,29 @@ public class Scheduler {
             }
         } catch (IOException e) {
             throw new IOError(e);
-        } finally {
-            lock.lock();
         }
 
         logger.debug("select over");
 
         // we use a set so that we don't execute any task twice
-        Collection<TaskConfiguration> executableTasks = new HashSet<TaskConfiguration>();
+        Collection<TaskIdentifier> executableTasks = new HashSet<TaskIdentifier>();
         for (SelectionKey sk : selector.selectedKeys()) {
-            TaskConfiguration[] subscribers = (TaskConfiguration[]) sk.attachment();
-
-            if (sk.isReadable()) {
-                addSubscriberTask(executableTasks, subscribers, EVENT_READ);
-            }
-            if (sk.isWritable()) {
-                addSubscriberTask(executableTasks, subscribers, EVENT_WRITE);
-            }
-            if (sk.isAcceptable()) {
-                addSubscriberTask(executableTasks, subscribers, EVENT_ACCEPT);
-            }
-            if (sk.isConnectable()) {
-                addSubscriberTask(executableTasks, subscribers, EVENT_CONNECT);
+            @SuppressWarnings("unchecked")
+            LinkedList<TaskInterestOps> subscribers = (LinkedList<TaskInterestOps>) sk.attachment();
+            for (TaskInterestOps ops : subscribers) {
+                if ((sk.readyOps() & ops.interestOps) != 0) {
+                    executableTasks.add(ops.tid);
+                    addReasonsFromInterestOp(ops.tid.ctx.reasons, sk.readyOps() & ops.interestOps);
+                }
             }
-
         }
-        for (TaskConfiguration tt : executableTasks) {
+        for (TaskIdentifier tt : executableTasks) {
             // cancel subscriptions to other events, we can execute now!
             tt.deregister();
             queueReady(tt);
         }
     }
 
-
     /**
      * Initialize and run scheduler. This function will return when all tasks
      * have completed.
@@ -529,10 +533,10 @@ public class Scheduler {
      */
     public static void run(Task initialTask) {
         logger.debug("running scheduler");
-        if (scheduler_running) {
+        if (schedulerRunning) {
             throw new AssertionError("Scheduler already running");
         }
-        scheduler_running = true;
+        schedulerRunning = true;
         try {
             run_unchecked(initialTask);
         } finally {
@@ -543,7 +547,6 @@ public class Scheduler {
         }
     }
 
-
     /**
      * Initialize and run scheduler. This function will return when all tasks
      * have completed. Don't check if the scheduler is already running or not.
@@ -556,12 +559,7 @@ public class Scheduler {
         }
 
         // the gnunet main loop
-        while (true) {
-            lock.lock();
-            if (checkLiveness() == false) {
-                lock.unlock();
-                break;
-            }
+        while (checkLiveness()) {
             RelativeTime nextTimeout = handleTimeouts();
             if (nextTimeout.getMicroseconds() < 0) {
                 logger.warn("negative timeout for select");
@@ -569,7 +567,6 @@ public class Scheduler {
 
             // don't select if there are no tasks; we are done!
             if (readyCount == 0 && pending.isEmpty()) {
-                lock.unlock();
                 return;
             }
 
@@ -580,7 +577,6 @@ public class Scheduler {
                 handleSelect(nextTimeout);
             }
             runReady();
-            lock.unlock();
         }
 
         if (readyCount != 0) {
@@ -618,9 +614,9 @@ public class Scheduler {
             // start executing from the highest priority down to 0
             for (int p = Priority.numberOfPriorities - 1; p >= 0; p--) {
                 // execute all tasks with priority p
-                LinkedList<TaskConfiguration> queue = readyLists[p];
+                LinkedList<TaskIdentifier> queue = readyLists[p];
                 while (!queue.isEmpty()) {
-                    TaskConfiguration tid = queue.removeFirst();
+                    TaskIdentifier tid = queue.removeFirst();
                     readyCount--;
                     tid.run();
                 }
@@ -637,13 +633,27 @@ public class Scheduler {
      */
     public static void shutdown() {
         // queueReady() while iterating would yield concurrent modification exn otherwise
-        for (TaskConfiguration tid : new ArrayList<TaskConfiguration>(pending)) {
+        for (TaskIdentifier tid : new ArrayList<TaskIdentifier>(pending)) {
             tid.ctx.reasons.add(Reason.SHUTDOWN);
             queueReady(tid);
         }
         pending.clear();
     }
 
+    /**
+     * Reset the scheduler forcefully.
+     * Intended to be used internally in the Scheduler, as well as in test teardown.
+     */
+    public static void forceReset() {
+        schedulerRunning = false;
+        readyCount = 0;
+        activeTask = null;
+        for (int i = 0; i < Priority.numberOfPriorities; ++i) {
+            readyLists[i] = Lists.newLinkedList();
+        }
+        pending.clear();
+    }
+
 
     /**
      * A handle to a file system object that can be selected on.