Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Commit the initial changes again - this time the diffs will show the …

…exact changes over the Scala code.
  • Loading branch information...
commit 1a3d112caa0c88041843277130198c3ef2f43e51 1 parent 3d94d77
@szegedi szegedi authored
Showing with 18 additions and 2,339 deletions.
  1. +1 −0  src/main/java/com/twitter/actors/threadpool/AbstractExecutorService.java
  2. +0 −342 src/main/java/com/twitter/actors/threadpool/BlockingQueue.java
  3. +1 −0  src/main/java/com/twitter/actors/threadpool/CompletionService.java
  4. +1 −1  src/main/java/com/twitter/actors/threadpool/ExecutorCompletionService.java
  5. +1 −1  src/main/java/com/twitter/actors/threadpool/ExecutorService.java
  6. +1 −1  src/main/java/com/twitter/actors/threadpool/Executors.java
  7. +1 −0  src/main/java/com/twitter/actors/threadpool/Future.java
  8. +1 −0  src/main/java/com/twitter/actors/threadpool/FutureTask.java
  9. +0 −751 src/main/java/com/twitter/actors/threadpool/LinkedBlockingQueue.java
  10. +0 −833 src/main/java/com/twitter/actors/threadpool/SynchronousQueue.java
  11. +1 −1  src/main/java/com/twitter/actors/threadpool/ThreadPoolExecutor.java
  12. +0 −407 src/main/java/com/twitter/actors/threadpool/TimeUnit.java
  13. +1 −0  src/main/java/com/twitter/actors/threadpool/helpers/Utils.java
  14. +1 −0  src/main/java/com/twitter/actors/threadpool/helpers/WaitQueue.java
  15. +1 −0  src/main/java/com/twitter/actors/threadpool/locks/CondVar.java
  16. +1 −0  src/main/java/com/twitter/actors/threadpool/locks/Condition.java
  17. +1 −0  src/main/java/com/twitter/actors/threadpool/locks/FIFOCondVar.java
  18. +1 −1  src/main/java/com/twitter/actors/threadpool/locks/Lock.java
  19. +1 −0  src/main/java/com/twitter/actors/threadpool/locks/ReentrantLock.java
  20. +1 −0  src/main/java/com/twitter/actors/threadpool/locks/ReentrantReadWriteLock.java
  21. +2 −1  src/main/scala/com/twitter/actors/FJTaskScheduler2.scala
View
1  src/main/java/com/twitter/actors/threadpool/AbstractExecutorService.java
@@ -11,6 +11,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Iterator;
+import java.util.concurrent.*;
/**
* Provides default implementations of {@link ExecutorService}
View
342 src/main/java/com/twitter/actors/threadpool/BlockingQueue.java
@@ -1,342 +0,0 @@
-/*
- * Written by Doug Lea with assistance from members of JCP JSR-166
- * Expert Group and released to the public domain, as explained at
- * http://creativecommons.org/licenses/publicdomain
- */
-
-package com.twitter.actors.threadpool;
-
-import java.util.Collection;
-
-/**
- * A {@link edu.emory.mathcs.backport.java.util.Queue} that additionally supports operations
- * that wait for the queue to become non-empty when retrieving an
- * element, and wait for space to become available in the queue when
- * storing an element.
- *
- * <p><tt>BlockingQueue</tt> methods come in four forms, with different ways
- * of handling operations that cannot be satisfied immediately, but may be
- * satisfied at some point in the future:
- * one throws an exception, the second returns a special value (either
- * <tt>null</tt> or <tt>false</tt>, depending on the operation), the third
- * blocks the current thread indefinitely until the operation can succeed,
- * and the fourth blocks for only a given maximum time limit before giving
- * up. These methods are summarized in the following table:
- *
- * <p>
- * <table BORDER CELLPADDING=3 CELLSPACING=1>
- * <tr>
- * <td></td>
- * <td ALIGN=CENTER><em>Throws exception</em></td>
- * <td ALIGN=CENTER><em>Special value</em></td>
- * <td ALIGN=CENTER><em>Blocks</em></td>
- * <td ALIGN=CENTER><em>Times out</em></td>
- * </tr>
- * <tr>
- * <td><b>Insert</b></td>
- * <td>{@link #add add(e)}</td>
- * <td>{@link #offer offer(e)}</td>
- * <td>{@link #put put(e)}</td>
- * <td>{@link #offer(Object, long, TimeUnit) offer(e, time, unit)}</td>
- * </tr>
- * <tr>
- * <td><b>Remove</b></td>
- * <td>{@link #remove remove()}</td>
- * <td>{@link #poll poll()}</td>
- * <td>{@link #take take()}</td>
- * <td>{@link #poll(long, TimeUnit) poll(time, unit)}</td>
- * </tr>
- * <tr>
- * <td><b>Examine</b></td>
- * <td>{@link #element element()}</td>
- * <td>{@link #peek peek()}</td>
- * <td><em>not applicable</em></td>
- * <td><em>not applicable</em></td>
- * </tr>
- * </table>
- *
- * <p>A <tt>BlockingQueue</tt> does not accept <tt>null</tt> elements.
- * Implementations throw <tt>NullPointerException</tt> on attempts
- * to <tt>add</tt>, <tt>put</tt> or <tt>offer</tt> a <tt>null</tt>. A
- * <tt>null</tt> is used as a sentinel value to indicate failure of
- * <tt>poll</tt> operations.
- *
- * <p>A <tt>BlockingQueue</tt> may be capacity bounded. At any given
- * time it may have a <tt>remainingCapacity</tt> beyond which no
- * additional elements can be <tt>put</tt> without blocking.
- * A <tt>BlockingQueue</tt> without any intrinsic capacity constraints always
- * reports a remaining capacity of <tt>Integer.MAX_VALUE</tt>.
- *
- * <p> <tt>BlockingQueue</tt> implementations are designed to be used
- * primarily for producer-consumer queues, but additionally support
- * the {@link java.util.Collection} interface. So, for example, it is
- * possible to remove an arbitrary element from a queue using
- * <tt>remove(x)</tt>. However, such operations are in general
- * <em>not</em> performed very efficiently, and are intended for only
- * occasional use, such as when a queued message is cancelled.
- *
- * <p> <tt>BlockingQueue</tt> implementations are thread-safe. All
- * queuing methods achieve their effects atomically using internal
- * locks or other forms of concurrency control. However, the
- * <em>bulk</em> Collection operations <tt>addAll</tt>,
- * <tt>containsAll</tt>, <tt>retainAll</tt> and <tt>removeAll</tt> are
- * <em>not</em> necessarily performed atomically unless specified
- * otherwise in an implementation. So it is possible, for example, for
- * <tt>addAll(c)</tt> to fail (throwing an exception) after adding
- * only some of the elements in <tt>c</tt>.
- *
- * <p>A <tt>BlockingQueue</tt> does <em>not</em> intrinsically support
- * any kind of &quot;close&quot; or &quot;shutdown&quot; operation to
- * indicate that no more items will be added. The needs and usage of
- * such features tend to be implementation-dependent. For example, a
- * common tactic is for producers to insert special
- * <em>end-of-stream</em> or <em>poison</em> objects, that are
- * interpreted accordingly when taken by consumers.
- *
- * <p>
- * Usage example, based on a typical producer-consumer scenario.
- * Note that a <tt>BlockingQueue</tt> can safely be used with multiple
- * producers and multiple consumers.
- * <pre>
- * class Producer implements Runnable {
- * private final BlockingQueue queue;
- * Producer(BlockingQueue q) { queue = q; }
- * public void run() {
- * try {
- * while (true) { queue.put(produce()); }
- * } catch (InterruptedException ex) { ... handle ...}
- * }
- * Object produce() { ... }
- * }
- *
- * class Consumer implements Runnable {
- * private final BlockingQueue queue;
- * Consumer(BlockingQueue q) { queue = q; }
- * public void run() {
- * try {
- * while (true) { consume(queue.take()); }
- * } catch (InterruptedException ex) { ... handle ...}
- * }
- * void consume(Object x) { ... }
- * }
- *
- * class Setup {
- * void main() {
- * BlockingQueue q = new SomeQueueImplementation();
- * Producer p = new Producer(q);
- * Consumer c1 = new Consumer(q);
- * Consumer c2 = new Consumer(q);
- * new Thread(p).start();
- * new Thread(c1).start();
- * new Thread(c2).start();
- * }
- * }
- * </pre>
- *
- * <p>Memory consistency effects: As with other concurrent
- * collections, actions in a thread prior to placing an object into a
- * {@code BlockingQueue}
- * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
- * actions subsequent to the access or removal of that element from
- * the {@code BlockingQueue} in another thread.
- *
- * <p>This interface is a member of the
- * <a href="{@docRoot}/../technotes/guides/collections/index.html">
- * Java Collections Framework</a>.
- *
- * @since 1.5
- * @author Doug Lea
- */
-public interface BlockingQueue extends Queue {
- /**
- * Inserts the specified element into this queue if it is possible to do
- * so immediately without violating capacity restrictions, returning
- * <tt>true</tt> upon success and throwing an
- * <tt>IllegalStateException</tt> if no space is currently available.
- * When using a capacity-restricted queue, it is generally preferable to
- * use {@link #offer(Object) offer}.
- *
- * @param e the element to add
- * @return <tt>true</tt> (as specified by {@link java.util.Collection#add})
- * @throws IllegalStateException if the element cannot be added at this
- * time due to capacity restrictions
- * @throws ClassCastException if the class of the specified element
- * prevents it from being added to this queue
- * @throws NullPointerException if the specified element is null
- * @throws IllegalArgumentException if some property of the specified
- * element prevents it from being added to this queue
- */
- boolean add(Object e);
-
- /**
- * Inserts the specified element into this queue if it is possible to do
- * so immediately without violating capacity restrictions, returning
- * <tt>true</tt> upon success and <tt>false</tt> if no space is currently
- * available. When using a capacity-restricted queue, this method is
- * generally preferable to {@link #add}, which can fail to insert an
- * element only by throwing an exception.
- *
- * @param e the element to add
- * @return <tt>true</tt> if the element was added to this queue, else
- * <tt>false</tt>
- * @throws ClassCastException if the class of the specified element
- * prevents it from being added to this queue
- * @throws NullPointerException if the specified element is null
- * @throws IllegalArgumentException if some property of the specified
- * element prevents it from being added to this queue
- */
- boolean offer(Object e);
-
- /**
- * Inserts the specified element into this queue, waiting if necessary
- * for space to become available.
- *
- * @param e the element to add
- * @throws InterruptedException if interrupted while waiting
- * @throws ClassCastException if the class of the specified element
- * prevents it from being added to this queue
- * @throws NullPointerException if the specified element is null
- * @throws IllegalArgumentException if some property of the specified
- * element prevents it from being added to this queue
- */
- void put(Object e) throws InterruptedException;
-
- /**
- * Inserts the specified element into this queue, waiting up to the
- * specified wait time if necessary for space to become available.
- *
- * @param e the element to add
- * @param timeout how long to wait before giving up, in units of
- * <tt>unit</tt>
- * @param unit a <tt>TimeUnit</tt> determining how to interpret the
- * <tt>timeout</tt> parameter
- * @return <tt>true</tt> if successful, or <tt>false</tt> if
- * the specified waiting time elapses before space is available
- * @throws InterruptedException if interrupted while waiting
- * @throws ClassCastException if the class of the specified element
- * prevents it from being added to this queue
- * @throws NullPointerException if the specified element is null
- * @throws IllegalArgumentException if some property of the specified
- * element prevents it from being added to this queue
- */
- boolean offer(Object e, long timeout, TimeUnit unit)
- throws InterruptedException;
-
- /**
- * Retrieves and removes the head of this queue, waiting if necessary
- * until an element becomes available.
- *
- * @return the head of this queue
- * @throws InterruptedException if interrupted while waiting
- */
- Object take() throws InterruptedException;
-
- /**
- * Retrieves and removes the head of this queue, waiting up to the
- * specified wait time if necessary for an element to become available.
- *
- * @param timeout how long to wait before giving up, in units of
- * <tt>unit</tt>
- * @param unit a <tt>TimeUnit</tt> determining how to interpret the
- * <tt>timeout</tt> parameter
- * @return the head of this queue, or <tt>null</tt> if the
- * specified waiting time elapses before an element is available
- * @throws InterruptedException if interrupted while waiting
- */
- Object poll(long timeout, TimeUnit unit)
- throws InterruptedException;
-
- /**
- * Returns the number of additional elements that this queue can ideally
- * (in the absence of memory or resource constraints) accept without
- * blocking, or <tt>Integer.MAX_VALUE</tt> if there is no intrinsic
- * limit.
- *
- * <p>Note that you <em>cannot</em> always tell if an attempt to insert
- * an element will succeed by inspecting <tt>remainingCapacity</tt>
- * because it may be the case that another thread is about to
- * insert or remove an element.
- *
- * @return the remaining capacity
- */
- int remainingCapacity();
-
- /**
- * Removes a single instance of the specified element from this queue,
- * if it is present. More formally, removes an element <tt>e</tt> such
- * that <tt>o.equals(e)</tt>, if this queue contains one or more such
- * elements.
- * Returns <tt>true</tt> if this queue contained the specified element
- * (or equivalently, if this queue changed as a result of the call).
- *
- * @param o element to be removed from this queue, if present
- * @return <tt>true</tt> if this queue changed as a result of the call
- * @throws ClassCastException if the class of the specified element
- * is incompatible with this queue (optional)
- * @throws NullPointerException if the specified element is null (optional)
- */
- boolean remove(Object o);
-
- /**
- * Returns <tt>true</tt> if this queue contains the specified element.
- * More formally, returns <tt>true</tt> if and only if this queue contains
- * at least one element <tt>e</tt> such that <tt>o.equals(e)</tt>.
- *
- * @param o object to be checked for containment in this queue
- * @return <tt>true</tt> if this queue contains the specified element
- * @throws ClassCastException if the class of the specified element
- * is incompatible with this queue (optional)
- * @throws NullPointerException if the specified element is null (optional)
- */
- public boolean contains(Object o);
-
- /**
- * Removes all available elements from this queue and adds them
- * to the given collection. This operation may be more
- * efficient than repeatedly polling this queue. A failure
- * encountered while attempting to add elements to
- * collection <tt>c</tt> may result in elements being in neither,
- * either or both collections when the associated exception is
- * thrown. Attempts to drain a queue to itself result in
- * <tt>IllegalArgumentException</tt>. Further, the behavior of
- * this operation is undefined if the specified collection is
- * modified while the operation is in progress.
- *
- * @param c the collection to transfer elements into
- * @return the number of elements transferred
- * @throws UnsupportedOperationException if addition of elements
- * is not supported by the specified collection
- * @throws ClassCastException if the class of an element of this queue
- * prevents it from being added to the specified collection
- * @throws NullPointerException if the specified collection is null
- * @throws IllegalArgumentException if the specified collection is this
- * queue, or some property of an element of this queue prevents
- * it from being added to the specified collection
- */
- int drainTo(Collection c);
-
- /**
- * Removes at most the given number of available elements from
- * this queue and adds them to the given collection. A failure
- * encountered while attempting to add elements to
- * collection <tt>c</tt> may result in elements being in neither,
- * either or both collections when the associated exception is
- * thrown. Attempts to drain a queue to itself result in
- * <tt>IllegalArgumentException</tt>. Further, the behavior of
- * this operation is undefined if the specified collection is
- * modified while the operation is in progress.
- *
- * @param c the collection to transfer elements into
- * @param maxElements the maximum number of elements to transfer
- * @return the number of elements transferred
- * @throws UnsupportedOperationException if addition of elements
- * is not supported by the specified collection
- * @throws ClassCastException if the class of an element of this queue
- * prevents it from being added to the specified collection
- * @throws NullPointerException if the specified collection is null
- * @throws IllegalArgumentException if the specified collection is this
- * queue, or some property of an element of this queue prevents
- * it from being added to the specified collection
- */
- int drainTo(Collection c, int maxElements);
-}
View
1  src/main/java/com/twitter/actors/threadpool/CompletionService.java
@@ -5,6 +5,7 @@
*/
package com.twitter.actors.threadpool;
+import java.util.concurrent.*;
/**
* A service that decouples the production of new asynchronous tasks
View
2  src/main/java/com/twitter/actors/threadpool/ExecutorCompletionService.java
@@ -6,7 +6,7 @@
package com.twitter.actors.threadpool;
import com.twitter.actors.threadpool.*; // for javadoc (till 6280605 is fixed)
-
+import java.util.concurrent.*;
/**
* A {@link CompletionService} that uses a supplied {@link Executor}
* to execute tasks. This class arranges that submitted tasks are,
View
2  src/main/java/com/twitter/actors/threadpool/ExecutorService.java
@@ -9,7 +9,7 @@
import com.twitter.actors.threadpool.*; // for javadoc (till 6280605 is fixed)
import java.util.List;
import java.util.Collection;
-
+import java.util.concurrent.*;
/**
* An {@link Executor} that provides methods to manage termination and
* methods that can produce a {@link Future} for tracking progress of
View
2  src/main/java/com/twitter/actors/threadpool/Executors.java
@@ -13,7 +13,7 @@
import java.security.AccessControlException;
import java.util.List;
import java.util.Collection;
-
+import java.util.concurrent.*;
/**
* Factory and utility methods for {@link Executor}, {@link
* ExecutorService}, {@link ScheduledExecutorService}, {@link
View
1  src/main/java/com/twitter/actors/threadpool/Future.java
@@ -6,6 +6,7 @@
package com.twitter.actors.threadpool;
import com.twitter.actors.threadpool.*; // for javadoc (till 6280605 is fixed)
+import java.util.concurrent.TimeUnit;
/**
* A <tt>Future</tt> represents the result of an asynchronous
View
1  src/main/java/com/twitter/actors/threadpool/FutureTask.java
@@ -8,6 +8,7 @@
import com.twitter.actors.threadpool.*; // for javadoc
import com.twitter.actors.threadpool.helpers.*;
+import java.util.concurrent.*;
/**
* A cancellable asynchronous computation. This class provides a base
View
751 src/main/java/com/twitter/actors/threadpool/LinkedBlockingQueue.java
@@ -1,751 +0,0 @@
-/*
- * Written by Doug Lea with assistance from members of JCP JSR-166
- * Expert Group and released to the public domain, as explained at
- * http://creativecommons.org/licenses/publicdomain
- */
-
-package com.twitter.actors.threadpool;
-
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-//import edu.emory.mathcs.backport.java.util.*;
-import com.twitter.actors.threadpool.helpers.*;
-
-/**
- * An optionally-bounded {@linkplain BlockingQueue blocking queue} based on
- * linked nodes.
- * This queue orders elements FIFO (first-in-first-out).
- * The <em>head</em> of the queue is that element that has been on the
- * queue the longest time.
- * The <em>tail</em> of the queue is that element that has been on the
- * queue the shortest time. New elements
- * are inserted at the tail of the queue, and the queue retrieval
- * operations obtain elements at the head of the queue.
- * Linked queues typically have higher throughput than array-based queues but
- * less predictable performance in most concurrent applications.
- *
- * <p> The optional capacity bound constructor argument serves as a
- * way to prevent excessive queue expansion. The capacity, if unspecified,
- * is equal to {@link Integer#MAX_VALUE}. Linked nodes are
- * dynamically created upon each insertion unless this would bring the
- * queue above capacity.
- *
- * <p>This class and its iterator implement all of the
- * <em>optional</em> methods of the {@link Collection} and {@link
- * Iterator} interfaces.
- *
- * <p>This class is a member of the
- * <a href="{@docRoot}/../technotes/guides/collections/index.html">
- * Java Collections Framework</a>.
- *
- * @since 1.5
- * @author Doug Lea
- *
- */
-public class LinkedBlockingQueue extends AbstractQueue
- implements BlockingQueue, java.io.Serializable {
- private static final long serialVersionUID = -6903933977591709194L;
-
- /*
- * A variant of the "two lock queue" algorithm. The putLock gates
- * entry to put (and offer), and has an associated condition for
- * waiting puts. Similarly for the takeLock. The "count" field
- * that they both rely on is maintained as an atomic to avoid
- * needing to get both locks in most cases. Also, to minimize need
- * for puts to get takeLock and vice-versa, cascading notifies are
- * used. When a put notices that it has enabled at least one take,
- * it signals taker. That taker in turn signals others if more
- * items have been entered since the signal. And symmetrically for
- * takes signalling puts. Operations such as remove(Object) and
- * iterators acquire both locks.
- */
-
- /**
- * Linked list node class
- */
- static class Node {
- /** The item, volatile to ensure barrier separating write and read */
- volatile Object item;
- Node next;
- Node(Object x) { item = x; }
- }
-
- /** The capacity bound, or Integer.MAX_VALUE if none */
- private final int capacity;
-
- /** Current number of elements */
- private volatile int count = 0;
-
- /** Head of linked list */
- private transient Node head;
-
- /** Tail of linked list */
- private transient Node last;
-
- /** Lock held by take, poll, etc */
- private final Object takeLock = new SerializableLock();
-
- /** Lock held by put, offer, etc */
- private final Object putLock = new SerializableLock();
-
- /**
- * Signals a waiting take. Called only from put/offer (which do not
- * otherwise ordinarily lock takeLock.)
- */
- private void signalNotEmpty() {
- synchronized (takeLock) {
- takeLock.notify();
- }
- }
-
- /**
- * Signals a waiting put. Called only from take/poll.
- */
- private void signalNotFull() {
- synchronized (putLock) {
- putLock.notify();
- }
- }
-
- /**
- * Creates a node and links it at end of queue.
- * @param x the item
- */
- private void insert(Object x) {
- last = last.next = new Node(x);
- }
-
- /**
- * Removes a node from head of queue,
- * @return the node
- */
- private Object extract() {
- Node first = head.next;
- head = first;
- Object x = first.item;
- first.item = null;
- return x;
- }
-
-
- /**
- * Creates a <tt>LinkedBlockingQueue</tt> with a capacity of
- * {@link Integer#MAX_VALUE}.
- */
- public LinkedBlockingQueue() {
- this(Integer.MAX_VALUE);
- }
-
- /**
- * Creates a <tt>LinkedBlockingQueue</tt> with the given (fixed) capacity.
- *
- * @param capacity the capacity of this queue
- * @throws IllegalArgumentException if <tt>capacity</tt> is not greater
- * than zero
- */
- public LinkedBlockingQueue(int capacity) {
- if (capacity <= 0) throw new IllegalArgumentException();
- this.capacity = capacity;
- last = head = new Node(null);
- }
-
- /**
- * Creates a <tt>LinkedBlockingQueue</tt> with a capacity of
- * {@link Integer#MAX_VALUE}, initially containing the elements of the
- * given collection,
- * added in traversal order of the collection's iterator.
- *
- * @param c the collection of elements to initially contain
- * @throws NullPointerException if the specified collection or any
- * of its elements are null
- */
- public LinkedBlockingQueue(Collection c) {
- this(Integer.MAX_VALUE);
- for (Iterator itr = c.iterator(); itr.hasNext();) {
- Object e = itr.next();
- add(e);
- }
- }
-
-
- // this doc comment is overridden to remove the reference to collections
- // greater in size than Integer.MAX_VALUE
- /**
- * Returns the number of elements in this queue.
- *
- * @return the number of elements in this queue
- */
- public int size() {
- return count;
- }
-
- // this doc comment is a modified copy of the inherited doc comment,
- // without the reference to unlimited queues.
- /**
- * Returns the number of additional elements that this queue can ideally
- * (in the absence of memory or resource constraints) accept without
- * blocking. This is always equal to the initial capacity of this queue
- * less the current <tt>size</tt> of this queue.
- *
- * <p>Note that you <em>cannot</em> always tell if an attempt to insert
- * an element will succeed by inspecting <tt>remainingCapacity</tt>
- * because it may be the case that another thread is about to
- * insert or remove an element.
- */
- public int remainingCapacity() {
- return capacity - count;
- }
-
- /**
- * Inserts the specified element at the tail of this queue, waiting if
- * necessary for space to become available.
- *
- * @throws InterruptedException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- */
- public void put(Object e) throws InterruptedException {
- if (e == null) throw new NullPointerException();
- // Note: convention in all put/take/etc is to preset
- // local var holding count negative to indicate failure unless set.
- int c = -1;
- synchronized (putLock) {
- /*
- * Note that count is used in wait guard even though it is
- * not protected by lock. This works because count can
- * only decrease at this point (all other puts are shut
- * out by lock), and we (or some other waiting put) are
- * signalled if it ever changes from
- * capacity. Similarly for all other uses of count in
- * other wait guards.
- */
- try {
- while (count == capacity)
- putLock.wait();
- } catch (InterruptedException ie) {
- putLock.notify(); // propagate to a non-interrupted thread
- throw ie;
- }
- insert(e);
- synchronized (this) { c = count++; }
- if (c + 1 < capacity)
- putLock.notify();
- }
-
- if (c == 0)
- signalNotEmpty();
- }
-
- /**
- * Inserts the specified element at the tail of this queue, waiting if
- * necessary up to the specified wait time for space to become available.
- *
- * @return <tt>true</tt> if successful, or <tt>false</tt> if
- * the specified waiting time elapses before space is available.
- * @throws InterruptedException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- */
- public boolean offer(Object e, long timeout, TimeUnit unit)
- throws InterruptedException {
-
- if (e == null) throw new NullPointerException();
- long nanos = unit.toNanos(timeout);
- int c = -1;
- synchronized (putLock) {
- long deadline = Utils.nanoTime() + nanos;
- for (;;) {
- if (count < capacity) {
- insert(e);
- synchronized (this) { c = count++; }
- if (c + 1 < capacity)
- putLock.notify();
- break;
- }
- if (nanos <= 0)
- return false;
- try {
- TimeUnit.NANOSECONDS.timedWait(putLock, nanos);
- nanos = deadline - Utils.nanoTime();
- } catch (InterruptedException ie) {
- putLock.notify(); // propagate to a non-interrupted thread
- throw ie;
- }
- }
- }
- if (c == 0)
- signalNotEmpty();
- return true;
- }
-
- /**
- * Inserts the specified element at the tail of this queue if it is
- * possible to do so immediately without exceeding the queue's capacity,
- * returning <tt>true</tt> upon success and <tt>false</tt> if this queue
- * is full.
- * When using a capacity-restricted queue, this method is generally
- * preferable to method {@link BlockingQueue#add add}, which can fail to
- * insert an element only by throwing an exception.
- *
- * @throws NullPointerException if the specified element is null
- */
- public boolean offer(Object e) {
- if (e == null) throw new NullPointerException();
- if (count == capacity)
- return false;
- int c = -1;
- synchronized (putLock) {
- if (count < capacity) {
- insert(e);
- synchronized (this) { c = count++; }
- if (c + 1 < capacity)
- putLock.notify();
- }
- }
- if (c == 0)
- signalNotEmpty();
- return c >= 0;
- }
-
-
- public Object take() throws InterruptedException {
- Object x;
- int c = -1;
- synchronized (takeLock) {
- try {
- while (count == 0)
- takeLock.wait();
- } catch (InterruptedException ie) {
- takeLock.notify(); // propagate to a non-interrupted thread
- throw ie;
- }
-
- x = extract();
- synchronized (this) { c = count--; }
- if (c > 1)
- takeLock.notify();
- }
- if (c == capacity)
- signalNotFull();
- return x;
- }
-
- public Object poll(long timeout, TimeUnit unit) throws InterruptedException {
- Object x = null;
- int c = -1;
- long nanos = unit.toNanos(timeout);
- synchronized (takeLock) {
- long deadline = Utils.nanoTime() + nanos;
- for (;;) {
- if (count > 0) {
- x = extract();
- synchronized (this) { c = count--; }
- if (c > 1)
- takeLock.notify();
- break;
- }
- if (nanos <= 0)
- return null;
- try {
- TimeUnit.NANOSECONDS.timedWait(takeLock, nanos);
- nanos = deadline - Utils.nanoTime();
- } catch (InterruptedException ie) {
- takeLock.notify(); // propagate to a non-interrupted thread
- throw ie;
- }
- }
- }
- if (c == capacity)
- signalNotFull();
- return x;
- }
-
- public Object poll() {
- if (count == 0)
- return null;
- Object x = null;
- int c = -1;
- synchronized (takeLock) {
- if (count > 0) {
- x = extract();
- synchronized (this) { c = count--; }
- if (c > 1)
- takeLock.notify();
- }
- }
- if (c == capacity)
- signalNotFull();
- return x;
- }
-
-
- public Object peek() {
- if (count == 0)
- return null;
- synchronized (takeLock) {
- Node first = head.next;
- if (first == null)
- return null;
- else
- return first.item;
- }
- }
-
- /**
- * Removes a single instance of the specified element from this queue,
- * if it is present. More formally, removes an element <tt>e</tt> such
- * that <tt>o.equals(e)</tt>, if this queue contains one or more such
- * elements.
- * Returns <tt>true</tt> if this queue contained the specified element
- * (or equivalently, if this queue changed as a result of the call).
- *
- * @param o element to be removed from this queue, if present
- * @return <tt>true</tt> if this queue changed as a result of the call
- */
- public boolean remove(Object o) {
- if (o == null) return false;
- boolean removed = false;
- synchronized (putLock) {
- synchronized (takeLock) {
- Node trail = head;
- Node p = head.next;
- while (p != null) {
- if (o.equals(p.item)) {
- removed = true;
- break;
- }
- trail = p;
- p = p.next;
- }
- if (removed) {
- p.item = null;
- trail.next = p.next;
- if (last == p)
- last = trail;
- synchronized (this) {
- if (count-- == capacity)
- putLock.notifyAll();
- }
- }
- }
- }
- return removed;
- }
-
- /**
- * Returns an array containing all of the elements in this queue, in
- * proper sequence.
- *
- * <p>The returned array will be "safe" in that no references to it are
- * maintained by this queue. (In other words, this method must allocate
- * a new array). The caller is thus free to modify the returned array.
- *
- * <p>This method acts as bridge between array-based and collection-based
- * APIs.
- *
- * @return an array containing all of the elements in this queue
- */
- public Object[] toArray() {
- synchronized (putLock) {
- synchronized (takeLock) {
- int size = count;
- Object[] a = new Object[size];
- int k = 0;
- for (Node p = head.next; p != null; p = p.next)
- a[k++] = p.item;
- return a;
- }
- }
- }
-
- /**
- * Returns an array containing all of the elements in this queue, in
- * proper sequence; the runtime type of the returned array is that of
- * the specified array. If the queue fits in the specified array, it
- * is returned therein. Otherwise, a new array is allocated with the
- * runtime type of the specified array and the size of this queue.
- *
- * <p>If this queue fits in the specified array with room to spare
- * (i.e., the array has more elements than this queue), the element in
- * the array immediately following the end of the queue is set to
- * <tt>null</tt>.
- *
- * <p>Like the {@link #toArray()} method, this method acts as bridge between
- * array-based and collection-based APIs. Further, this method allows
- * precise control over the runtime type of the output array, and may,
- * under certain circumstances, be used to save allocation costs.
- *
- * <p>Suppose <tt>x</tt> is a queue known to contain only strings.
- * The following code can be used to dump the queue into a newly
- * allocated array of <tt>String</tt>:
- *
- * <pre>
- * String[] y = x.toArray(new String[0]);</pre>
- *
- * Note that <tt>toArray(new Object[0])</tt> is identical in function to
- * <tt>toArray()</tt>.
- *
- * @param a the array into which the elements of the queue are to
- * be stored, if it is big enough; otherwise, a new array of the
- * same runtime type is allocated for this purpose
- * @return an array containing all of the elements in this queue
- * @throws ArrayStoreException if the runtime type of the specified array
- * is not a supertype of the runtime type of every element in
- * this queue
- * @throws NullPointerException if the specified array is null
- */
- public Object[] toArray(Object[] a) {
- synchronized (putLock) {
- synchronized (takeLock) {
- int size = count;
- if (a.length < size)
- a = (Object[])java.lang.reflect.Array.newInstance
- (a.getClass().getComponentType(), size);
-
- int k = 0;
- for (Node p = head.next; p != null; p = p.next)
- a[k++] = (Object)p.item;
- if (a.length > k)
- a[k] = null;
- return a;
- }
- }
- }
-
- public String toString() {
- synchronized (putLock) {
- synchronized (takeLock) {
- return super.toString();
- }
- }
- }
-
- /**
- * Atomically removes all of the elements from this queue.
- * The queue will be empty after this call returns.
- */
- public void clear() {
- synchronized (putLock) {
- synchronized (takeLock) {
- head.next = null;
- assert head.item == null;
- last = head;
- int c;
- synchronized (this) {
- c = count;
- count = 0;
- }
- if (c == capacity)
- putLock.notifyAll();
- }
- }
- }
-
- /**
- * @throws UnsupportedOperationException {@inheritDoc}
- * @throws ClassCastException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- * @throws IllegalArgumentException {@inheritDoc}
- */
- public int drainTo(Collection c) {
- if (c == null)
- throw new NullPointerException();
- if (c == this)
- throw new IllegalArgumentException();
- Node first;
- synchronized (putLock) {
- synchronized (takeLock) {
- first = head.next;
- head.next = null;
- assert head.item == null;
- last = head;
- int cold;
- synchronized (this) {
- cold = count;
- count = 0;
- }
- if (cold == capacity)
- putLock.notifyAll();
- }
- }
- // Transfer the elements outside of locks
- int n = 0;
- for (Node p = first; p != null; p = p.next) {
- c.add(p.item);
- p.item = null;
- ++n;
- }
- return n;
- }
-
- /**
- * @throws UnsupportedOperationException {@inheritDoc}
- * @throws ClassCastException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- * @throws IllegalArgumentException {@inheritDoc}
- */
- public int drainTo(Collection c, int maxElements) {
- if (c == null)
- throw new NullPointerException();
- if (c == this)
- throw new IllegalArgumentException();
- synchronized (putLock) {
- synchronized (takeLock) {
- int n = 0;
- Node p = head.next;
- while (p != null && n < maxElements) {
- c.add(p.item);
- p.item = null;
- p = p.next;
- ++n;
- }
- if (n != 0) {
- head.next = p;
- assert head.item == null;
- if (p == null)
- last = head;
- int cold;
- synchronized (this) {
- cold = count;
- count -= n;
- }
- if (cold == capacity)
- putLock.notifyAll();
- }
- return n;
- }
- }
- }
-
- /**
- * Returns an iterator over the elements in this queue in proper sequence.
- * The returned <tt>Iterator</tt> is a "weakly consistent" iterator that
- * will never throw {@link java.util.ConcurrentModificationException},
- * and guarantees to traverse elements as they existed upon
- * construction of the iterator, and may (but is not guaranteed to)
- * reflect any modifications subsequent to construction.
- *
- * @return an iterator over the elements in this queue in proper sequence
- */
- public Iterator iterator() {
- return new Itr();
- }
-
- private class Itr implements Iterator {
- /*
- * Basic weak-consistent iterator. At all times hold the next
- * item to hand out so that if hasNext() reports true, we will
- * still have it to return even if lost race with a take etc.
- */
- private Node current;
- private Node lastRet;
- private Object currentElement;
-
- Itr() {
- synchronized (putLock) {
- synchronized (takeLock) {
- current = head.next;
- if (current != null)
- currentElement = current.item;
- }
- }
- }
-
- public boolean hasNext() {
- return current != null;
- }
-
- public Object next() {
- synchronized (putLock) {
- synchronized (takeLock) {
- if (current == null)
- throw new NoSuchElementException();
- Object x = currentElement;
- lastRet = current;
- current = current.next;
- if (current != null)
- currentElement = current.item;
- return x;
- }
- }
- }
-
- public void remove() {
- if (lastRet == null)
- throw new IllegalStateException();
- synchronized (putLock) {
- synchronized (takeLock) {
- Node node = lastRet;
- lastRet = null;
- Node trail = head;
- Node p = head.next;
- while (p != null && p != node) {
- trail = p;
- p = p.next;
- }
- if (p == node) {
- p.item = null;
- trail.next = p.next;
- if (last == p)
- last = trail;
- int c;
- synchronized (this) { c = count--; }
- if (c == capacity)
- putLock.notifyAll();
- }
- }
- }
- }
- }
-
- /**
- * Save the state to a stream (that is, serialize it).
- *
- * @serialData The capacity is emitted (int), followed by all of
- * its elements (each an <tt>Object</tt>) in the proper order,
- * followed by a null
- * @param s the stream
- */
- private void writeObject(java.io.ObjectOutputStream s)
- throws java.io.IOException {
-
- synchronized (putLock) {
- synchronized (takeLock) {
- // Write out any hidden stuff, plus capacity
- s.defaultWriteObject();
-
- // Write out all elements in the proper order.
- for (Node p = head.next; p != null; p = p.next)
- s.writeObject(p.item);
-
- // Use trailing null as sentinel
- s.writeObject(null);
- }
- }
- }
-
- /**
- * Reconstitute this queue instance from a stream (that is,
- * deserialize it).
- * @param s the stream
- */
- private void readObject(java.io.ObjectInputStream s)
- throws java.io.IOException, ClassNotFoundException {
- // Read in capacity, and any hidden stuff
- s.defaultReadObject();
-
- synchronized (this) { count = 0; }
- last = head = new Node(null);
-
- // Read in all elements and place in queue
- for (;;) {
- Object item = (Object)s.readObject();
- if (item == null)
- break;
- add(item);
- }
- }
-
- private static class SerializableLock implements java.io.Serializable {
- private final static long serialVersionUID = -8856990691138858668L;
- }
-}
View
833 src/main/java/com/twitter/actors/threadpool/SynchronousQueue.java
@@ -1,833 +0,0 @@
-/*
- * Written by Doug Lea with assistance from members of JCP JSR-166
- * Expert Group and released to the public domain, as explained at
- * http://creativecommons.org/licenses/publicdomain
- */
-
-package com.twitter.actors.threadpool;
-import com.twitter.actors.threadpool.locks.*;
-//import edu.emory.mathcs.backport.java.util.*;
-import java.util.Collection;
-import java.util.Iterator;
-import com.twitter.actors.threadpool.helpers.Utils;
-import java.util.NoSuchElementException;
-
-/**
- * A {@linkplain BlockingQueue blocking queue} in which each insert
- * operation must wait for a corresponding remove operation by another
- * thread, and vice versa. A synchronous queue does not have any
- * internal capacity, not even a capacity of one. You cannot
- * <tt>peek</tt> at a synchronous queue because an element is only
- * present when you try to remove it; you cannot insert an element
- * (using any method) unless another thread is trying to remove it;
- * you cannot iterate as there is nothing to iterate. The
- * <em>head</em> of the queue is the element that the first queued
- * inserting thread is trying to add to the queue; if there is no such
- * queued thread then no element is available for removal and
- * <tt>poll()</tt> will return <tt>null</tt>. For purposes of other
- * <tt>Collection</tt> methods (for example <tt>contains</tt>), a
- * <tt>SynchronousQueue</tt> acts as an empty collection. This queue
- * does not permit <tt>null</tt> elements.
- *
- * <p>Synchronous queues are similar to rendezvous channels used in
- * CSP and Ada. They are well suited for handoff designs, in which an
- * object running in one thread must sync up with an object running
- * in another thread in order to hand it some information, event, or
- * task.
- *
- * <p> This class supports an optional fairness policy for ordering
- * waiting producer and consumer threads. By default, this ordering
- * is not guaranteed. However, a queue constructed with fairness set
- * to <tt>true</tt> grants threads access in FIFO order. Fairness
- * generally decreases throughput but reduces variability and avoids
- * starvation.
- *
- * <p>This class and its iterator implement all of the
- * <em>optional</em> methods of the {@link Collection} and {@link
- * Iterator} interfaces.
- *
- * <p>This class is a member of the
- * <a href="{@docRoot}/../technotes/guides/collections/index.html">
- * Java Collections Framework</a>.
- *
- * @since 1.5
- * @author Doug Lea
- */
-public class SynchronousQueue extends AbstractQueue
- implements BlockingQueue, java.io.Serializable {
- private static final long serialVersionUID = -3223113410248163686L;
-
- /*
- This implementation divides actions into two cases for puts:
-
- * An arriving producer that does not already have a waiting consumer
- creates a node holding item, and then waits for a consumer to take it.
- * An arriving producer that does already have a waiting consumer fills
- the slot node created by the consumer, and notifies it to continue.
-
- And symmetrically, two for takes:
-
- * An arriving consumer that does not already have a waiting producer
- creates an empty slot node, and then waits for a producer to fill it.
- * An arriving consumer that does already have a waiting producer takes
- item from the node created by the producer, and notifies it to continue.
-
- When a put or take waiting for the actions of its counterpart
- aborts due to interruption or timeout, it marks the node
- it created as "CANCELLED", which causes its counterpart to retry
- the entire put or take sequence.
-
- This requires keeping two simple queues, waitingProducers and
- waitingConsumers. Each of these can be FIFO (preserves fairness)
- or LIFO (improves throughput).
- */
-
- /** Lock protecting both wait queues */
- private final ReentrantLock qlock;
- /** Queue holding waiting puts */
- private final WaitQueue waitingProducers;
- /** Queue holding waiting takes */
- private final WaitQueue waitingConsumers;
-
- /**
- * Creates a <tt>SynchronousQueue</tt> with nonfair access policy.
- */
- public SynchronousQueue() {
- this(false);
- }
-
- /**
- * Creates a <tt>SynchronousQueue</tt> with specified fairness policy.
- * @param fair if true, threads contend in FIFO order for access;
- * otherwise the order is unspecified.
- */
- public SynchronousQueue(boolean fair) {
- if (fair) {
- qlock = new ReentrantLock(true);
- waitingProducers = new FifoWaitQueue();
- waitingConsumers = new FifoWaitQueue();
- }
- else {
- qlock = new ReentrantLock();
- waitingProducers = new LifoWaitQueue();
- waitingConsumers = new LifoWaitQueue();
- }
- }
-
- /**
- * Queue to hold waiting puts/takes; specialized to Fifo/Lifo below.
- * These queues have all transient fields, but are serializable
- * in order to recover fairness settings when deserialized.
- */
- static abstract class WaitQueue implements java.io.Serializable {
- /** Creates, adds, and returns node for x. */
- abstract Node enq(Object x);
- /** Removes and returns node, or null if empty. */
- abstract Node deq();
- /** Removes a cancelled node to avoid garbage retention. */
- abstract void unlink(Node node);
- /** Returns true if a cancelled node might be on queue. */
- abstract boolean shouldUnlink(Node node);
- }
-
- /**
- * FIFO queue to hold waiting puts/takes.
- */
- static final class FifoWaitQueue extends WaitQueue implements java.io.Serializable {
- private static final long serialVersionUID = -3623113410248163686L;
- private transient Node head;
- private transient Node last;
-
- Node enq(Object x) {
- Node p = new Node(x);
- if (last == null)
- last = head = p;
- else
- last = last.next = p;
- return p;
- }
-
- Node deq() {
- Node p = head;
- if (p != null) {
- if ((head = p.next) == null)
- last = null;
- p.next = null;
- }
- return p;
- }
-
- boolean shouldUnlink(Node node) {
- return (node == last || node.next != null);
- }
-
- void unlink(Node node) {
- Node p = head;
- Node trail = null;
- while (p != null) {
- if (p == node) {
- Node next = p.next;
- if (trail == null)
- head = next;
- else
- trail.next = next;
- if (last == node)
- last = trail;
- break;
- }
- trail = p;
- p = p.next;
- }
- }
- }
-
- /**
- * LIFO queue to hold waiting puts/takes.
- */
- static final class LifoWaitQueue extends WaitQueue implements java.io.Serializable {
- private static final long serialVersionUID = -3633113410248163686L;
- private transient Node head;
-
- Node enq(Object x) {
- return head = new Node(x, head);
- }
-
- Node deq() {
- Node p = head;
- if (p != null) {
- head = p.next;
- p.next = null;
- }
- return p;
- }
-
- boolean shouldUnlink(Node node) {
- // Return false if already dequeued or is bottom node (in which
- // case we might retain at most one garbage node)
- return (node == head || node.next != null);
- }
-
- void unlink(Node node) {
- Node p = head;
- Node trail = null;
- while (p != null) {
- if (p == node) {
- Node next = p.next;
- if (trail == null)
- head = next;
- else
- trail.next = next;
- break;
- }
- trail = p;
- p = p.next;
- }
- }
- }
-
- /**
- * Unlinks the given node from consumer queue. Called by cancelled
- * (timeout, interrupt) waiters to avoid garbage retention in the
- * absence of producers.
- */
- private void unlinkCancelledConsumer(Node node) {
- // Use a form of double-check to avoid unnecessary locking and
- // traversal. The first check outside lock might
- // conservatively report true.
- if (waitingConsumers.shouldUnlink(node)) {
- qlock.lock();
- try {
- if (waitingConsumers.shouldUnlink(node))
- waitingConsumers.unlink(node);
- } finally {
- qlock.unlock();
- }
- }
- }
-
- /**
- * Unlinks the given node from producer queue. Symmetric
- * to unlinkCancelledConsumer.
- */
- private void unlinkCancelledProducer(Node node) {
- if (waitingProducers.shouldUnlink(node)) {
- qlock.lock();
- try {
- if (waitingProducers.shouldUnlink(node))
- waitingProducers.unlink(node);
- } finally {
- qlock.unlock();
- }
- }
- }
-
- /**
- * Nodes each maintain an item and handle waits and signals for
- * getting and setting it. The class extends
- * AbstractQueuedSynchronizer to manage blocking, using AQS state
- * 0 for waiting, 1 for ack, -1 for cancelled.
- */
- static final class Node implements java.io.Serializable {
- private static final long serialVersionUID = -3223113410248163686L;
-
- /** Synchronization state value representing that node acked */
- private static final int ACK = 1;
- /** Synchronization state value representing that node cancelled */
- private static final int CANCEL = -1;
-
- int state = 0;
-
- /** The item being transferred */
- Object item;
- /** Next node in wait queue */
- Node next;
-
- /** Creates a node with initial item */
- Node(Object x) { item = x; }
-
- /** Creates a node with initial item and next */
- Node(Object x, Node n) { item = x; next = n; }
-
- /**
- * Takes item and nulls out field (for sake of GC)
- *
- * PRE: lock owned
- */
- private Object extract() {
- Object x = item;
- item = null;
- return x;
- }
-
- /**
- * Tries to cancel on interrupt; if so rethrowing,
- * else setting interrupt state
- *
- * PRE: lock owned
- */
- private void checkCancellationOnInterrupt(InterruptedException ie)
- throws InterruptedException
- {
- if (state == 0) {
- state = CANCEL;
- notify();
- throw ie;
- }
- Thread.currentThread().interrupt();
- }
-
- /**
- * Fills in the slot created by the consumer and signal consumer to
- * continue.
- */
- synchronized boolean setItem(Object x) {
- if (state != 0) return false;
- item = x;
- state = ACK;
- notify();
- return true;
- }
-
- /**
- * Removes item from slot created by producer and signal producer
- * to continue.
- */
- synchronized Object getItem() {
- if (state != 0) return null;
- state = ACK;
- notify();
- return extract();
- }
-
- /**
- * Waits for a consumer to take item placed by producer.
- */
- synchronized void waitForTake() throws InterruptedException {
- try {
- while (state == 0) wait();
- } catch (InterruptedException ie) {
- checkCancellationOnInterrupt(ie);
- }
- }
-
- /**
- * Waits for a producer to put item placed by consumer.
- */
- synchronized Object waitForPut() throws InterruptedException {
- try {
- while (state == 0) wait();
- } catch (InterruptedException ie) {
- checkCancellationOnInterrupt(ie);
- }
- return extract();
- }
-
- private boolean attempt(long nanos) throws InterruptedException {
- if (state != 0) return true;
- if (nanos <= 0) {
- state = CANCEL;
- notify();
- return false;
- }
- long deadline = Utils.nanoTime() + nanos;
- while (true) {
- TimeUnit.NANOSECONDS.timedWait(this, nanos);
- if (state != 0) return true;
- nanos = deadline - Utils.nanoTime();
- if (nanos <= 0) {
- state = CANCEL;
- notify();
- return false;
- }
- }
- }
-
- /**
- * Waits for a consumer to take item placed by producer or time out.
- */
- synchronized boolean waitForTake(long nanos) throws InterruptedException {
- try {
- if (!attempt(nanos)) return false;
- } catch (InterruptedException ie) {
- checkCancellationOnInterrupt(ie);
- }
- return true;
- }
-
- /**
- * Waits for a producer to put item placed by consumer, or time out.
- */
- synchronized Object waitForPut(long nanos) throws InterruptedException {
- try {
- if (!attempt(nanos)) return null;
- } catch (InterruptedException ie) {
- checkCancellationOnInterrupt(ie);
- }
- return extract();
- }
- }
-
- /**
- * Adds the specified element to this queue, waiting if necessary for
- * another thread to receive it.
- *
- * @throws InterruptedException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- */
- public void put(Object e) throws InterruptedException {
- if (e == null) throw new NullPointerException();
- final ReentrantLock qlock = this.qlock;
-
- for (;;) {
- Node node;
- boolean mustWait;
- if (Thread.interrupted()) throw new InterruptedException();
- qlock.lock();
- try {
- node = waitingConsumers.deq();
- if ( (mustWait = (node == null)) )
- node = waitingProducers.enq(e);
- } finally {
- qlock.unlock();
- }
-
- if (mustWait) {
- try {
- node.waitForTake();
- return;
- } catch (InterruptedException ex) {
- unlinkCancelledProducer(node);
- throw ex;
- }
- }
-
- else if (node.setItem(e))
- return;
-
- // else consumer cancelled, so retry
- }
- }
-
- /**
- * Inserts the specified element into this queue, waiting if necessary
- * up to the specified wait time for another thread to receive it.
- *
- * @return <tt>true</tt> if successful, or <tt>false</tt> if the
- * specified waiting time elapses before a consumer appears.
- * @throws InterruptedException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- */
- public boolean offer(Object e, long timeout, TimeUnit unit) throws InterruptedException {
- if (e == null) throw new NullPointerException();
- long nanos = unit.toNanos(timeout);
- final ReentrantLock qlock = this.qlock;
- for (;;) {
- Node node;
- boolean mustWait;
- if (Thread.interrupted()) throw new InterruptedException();
- qlock.lock();
- try {
- node = waitingConsumers.deq();
- if ( (mustWait = (node == null)) )
- node = waitingProducers.enq(e);
- } finally {
- qlock.unlock();
- }
-
- if (mustWait) {
- try {
- boolean x = node.waitForTake(nanos);
- if (!x)
- unlinkCancelledProducer(node);
- return x;
- } catch (InterruptedException ex) {
- unlinkCancelledProducer(node);
- throw ex;
- }
- }
-
- else if (node.setItem(e))
- return true;
-
- // else consumer cancelled, so retry
- }
- }
-
- /**
- * Retrieves and removes the head of this queue, waiting if necessary
- * for another thread to insert it.
- *
- * @return the head of this queue
- * @throws InterruptedException {@inheritDoc}
- */
- public Object take() throws InterruptedException {
- final ReentrantLock qlock = this.qlock;
- for (;;) {
- Node node;
- boolean mustWait;
-
- if (Thread.interrupted()) throw new InterruptedException();
- qlock.lock();
- try {
- node = waitingProducers.deq();
- if ( (mustWait = (node == null)) )
- node = waitingConsumers.enq(null);
- } finally {
- qlock.unlock();
- }
-
- if (mustWait) {
- try {
- Object x = node.waitForPut();
- return (Object)x;
- } catch (InterruptedException ex) {
- unlinkCancelledConsumer(node);
- throw ex;
- }
- }
- else {
- Object x = node.getItem();
- if (x != null)
- return (Object)x;
- // else cancelled, so retry
- }
- }
- }
-
- /**
- * Retrieves and removes the head of this queue, waiting
- * if necessary up to the specified wait time, for another thread
- * to insert it.
- *
- * @return the head of this queue, or <tt>null</tt> if the
- * specified waiting time elapses before an element is present.
- * @throws InterruptedException {@inheritDoc}
- */
- public Object poll(long timeout, TimeUnit unit) throws InterruptedException {
- long nanos = unit.toNanos(timeout);
- final ReentrantLock qlock = this.qlock;
-
- for (;;) {
- Node node;
- boolean mustWait;
-
- if (Thread.interrupted()) throw new InterruptedException();
- qlock.lock();
- try {
- node = waitingProducers.deq();
- if ( (mustWait = (node == null)) )
- node = waitingConsumers.enq(null);
- } finally {
- qlock.unlock();
- }
-
- if (mustWait) {
- try {
- Object x = node.waitForPut(nanos);
- if (x == null)
- unlinkCancelledConsumer(node);
- return (Object)x;
- } catch (InterruptedException ex) {
- unlinkCancelledConsumer(node);
- throw ex;
- }
- }
- else {
- Object x = node.getItem();
- if (x != null)
- return (Object)x;
- // else cancelled, so retry
- }
- }
- }
-
- // Untimed nonblocking versions
-
- /**
- * Inserts the specified element into this queue, if another thread is
- * waiting to receive it.
- *
- * @param e the element to add
- * @return <tt>true</tt> if the element was added to this queue, else
- * <tt>false</tt>
- * @throws NullPointerException if the specified element is null
- */
- public boolean offer(Object e) {
- if (e == null) throw new NullPointerException();
- final ReentrantLock qlock = this.qlock;
-
- for (;;) {
- Node node;
- qlock.lock();
- try {
- node = waitingConsumers.deq();
- } finally {
- qlock.unlock();
- }
- if (node == null)
- return false;
-
- else if (node.setItem(e))
- return true;
- // else retry
- }
- }
-
- /**
- * Retrieves and removes the head of this queue, if another thread
- * is currently making an element available.
- *
- * @return the head of this queue, or <tt>null</tt> if no
- * element is available.
- */
- public Object poll() {
- final ReentrantLock qlock = this.qlock;
- for (;;) {
- Node node;
- qlock.lock();
- try {
- node = waitingProducers.deq();
- } finally {
- qlock.unlock();
- }
- if (node == null)
- return null;
-
- else {
- Object x = node.getItem();
- if (x != null)
- return (Object)x;
- // else retry
- }
- }
- }
-
- /**
- * Always returns <tt>true</tt>.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @return <tt>true</tt>
- */
- public boolean isEmpty() {
- return true;
- }
-
- /**
- * Always returns zero.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @return zero
- */
- public int size() {
- return 0;
- }
-
- /**
- * Always returns zero.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @return zero
- */
- public int remainingCapacity() {
- return 0;
- }
-
- /**
- * Does nothing.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- */
- public void clear() {}
-
- /**
- * Always returns <tt>false</tt>.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @param o object to be checked for containment in this queue
- * @return <tt>false</tt>
- */
- public boolean contains(Object o) {
- return false;
- }
-
- /**
- * Always returns <tt>false</tt>.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @param o the element to remove
- * @return <tt>false</tt>
- */
- public boolean remove(Object o) {
- return false;
- }
-
- /**
- * Returns <tt>false</tt> unless the given collection is empty.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @param c the collection
- * @return <tt>false</tt> unless the given collection is empty
- * @throws NullPointerException if the specified collection is null
- */
- public boolean containsAll(Collection c) {
- return c.isEmpty();
- }
-
- /**
- * Always returns <tt>false</tt>.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @param c the collection
- * @return <tt>false</tt>
- */
- public boolean removeAll(Collection c) {
- return false;
- }
-
- /**
- * Always returns <tt>false</tt>.
- * A <tt>SynchronousQueue</tt> has no internal capacity.
- *
- * @param c the collection
- * @return <tt>false</tt>
- */
- public boolean retainAll(Collection c) {
- return false;
- }
-
- /**
- * Always returns <tt>null</tt>.
- * A <tt>SynchronousQueue</tt> does not return elements
- * unless actively waited on.
- *
- * @return <tt>null</tt>
- */
- public Object peek() {
- return null;
- }
-
-
- static class EmptyIterator implements Iterator {
- public boolean hasNext() {
- return false;
- }
- public Object next() {
- throw new NoSuchElementException();
- }
- public void remove() {
- throw new IllegalStateException();
- }
- }
-
- /**
- * Returns an empty iterator in which <tt>hasNext</tt> always returns
- * <tt>false</tt>.
- *
- * @return an empty iterator
- */
- public Iterator iterator() {
- return new EmptyIterator();
- }
-
-
- /**
- * Returns a zero-length array.
- * @return a zero-length array
- */
- public Object[] toArray() {
- return new Object[0];
- }
-
- /**
- * Sets the zeroeth element of the specified array to <tt>null</tt>
- * (if the array has non-zero length) and returns it.
- *
- * @param a the array
- * @return the specified array
- * @throws NullPointerException if the specified array is null
- */
- public Object[] toArray(Object[] a) {
- if (a.length > 0)
- a[0] = null;
- return a;
- }
-
- /**
- * @throws UnsupportedOperationException {@inheritDoc}
- * @throws ClassCastException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- * @throws IllegalArgumentException {@inheritDoc}
- */
- public int drainTo(Collection c) {
- if (c == null)
- throw new NullPointerException();
- if (c == this)
- throw new IllegalArgumentException();
- int n = 0;
- Object e;
- while ( (e = poll()) != null) {
- c.add(e);
- ++n;
- }
- return n;
- }
-
- /**
- * @throws UnsupportedOperationException {@inheritDoc}
- * @throws ClassCastException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- * @throws IllegalArgumentException {@inheritDoc}
- */
- public int drainTo(Collection c, int maxElements) {
- if (c == null)
- throw new NullPointerException();
- if (c == this)
- throw new IllegalArgumentException();
- int n = 0;
- Object e;
- while (n < maxElements && (e = poll()) != null) {
- c.add(e);
- ++n;
- }
- return n;
- }
-}
View
2  src/main/java/com/twitter/actors/threadpool/ThreadPoolExecutor.java
@@ -12,7 +12,7 @@
import java.util.Iterator;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
-
+import java.util.concurrent.*;
/**
* An {@link ExecutorService} that executes each submitted task using
* one of possibly several pooled threads, normally configured
View
407 src/main/java/com/twitter/actors/threadpool/TimeUnit.java
@@ -1,407 +0,0 @@
-/*
- * Written by Doug Lea with assistance from members of JCP JSR-166
- * Expert Group and released to the public domain, as explained at
- * http://creativecommons.org/licenses/publicdomain
- */
-
-package com.twitter.actors.threadpool;
-
-import java.io.InvalidObjectException;
-import java.io.ObjectStreamException;
-
-/**
- * A <tt>TimeUnit</tt> represents time durations at a given unit of
- * granularity and provides utility methods to convert across units,
- * and to perform timing and delay operations in these units. A
- * <tt>TimeUnit</tt> does not maintain time information, but only
- * helps organize and use time representations that may be maintained
- * separately across various contexts. A nanosecond is defined as one
- * thousandth of a microsecond, a microsecond as one thousandth of a
- * millisecond, a millisecond as one thousandth of a second, a minute
- * as sixty seconds, an hour as sixty minutes, and a day as twenty four
- * hours.
- *
- * <p>A <tt>TimeUnit</tt> is mainly used to inform time-based methods
- * how a given timing parameter should be interpreted. For example,
- * the following code will timeout in 50 milliseconds if the {@link
- * edu.emory.mathcs.backport.java.util.concurrent.locks.Lock lock} is not available:
- *
- * <pre> Lock lock = ...;
- * if ( lock.tryLock(50L, TimeUnit.MILLISECONDS) ) ...
- * </pre>
- * while this code will timeout in 50 seconds:
- * <pre>
- * Lock lock = ...;
- * if ( lock.tryLock(50L, TimeUnit.SECONDS) ) ...
- * </pre>
- *
- * Note however, that there is no guarantee that a particular timeout
- * implementation will be able to notice the passage of time at the
- * same granularity as the given <tt>TimeUnit</tt>.
- *
- * @since 1.5
- * @author Doug Lea
- */
-public abstract class TimeUnit implements java.io.Serializable {
-
- public static final TimeUnit NANOSECONDS = new TimeUnit(0, "NANOSECONDS") {
- private final static long serialVersionUID = 535148490883208361L;
- public long toNanos(long d) { return d; }
- public long toMicros(long d) { return d/(C1/C0); }
- public long toMillis(long d) { return d/(C2/C0); }
- public long toSeconds(long d) { return d/(C3/C0); }
- public long toMinutes(long d) { return d/(C4/C0); }
- public long toHours(long d) { return d/(C5/C0); }
- public long toDays(long d) { return d/(C6/C0); }
- public long convert(long d, TimeUnit u) { return u.toNanos(d); }
- int excessNanos(long d, long m) { return (int)(d - (m*C2)); }
- };
- public static final TimeUnit MICROSECONDS = new TimeUnit(1, "MICROSECONDS") {
- private final static long serialVersionUID = 2185906575929579108L;
- public long toNanos(long d) { return x(d, C1/C0, MAX/(C1/C0)); }
- public long toMicros(long d) { return d; }
- public long toMillis(long d) { return d/(C2/C1); }
- public long toSeconds(long d) { return d/(C3/C1); }
- public long toMinutes(long d) { return d/(C4/C1); }
- public long toHours(long d) { return d/(C5/C1); }
- public long toDays(long d) { return d/(C6/C1); }
- public long convert(long d, TimeUnit u) { return u.toMicros(d); }
- int excessNanos(long d, long m) { return (int)((d*C1) - (m*C2)); }
- };
- public static final TimeUnit MILLISECONDS = new TimeUnit(2, "MILLISECONDS") {
- private final static long serialVersionUID = 9032047794123325184L;
- public long toNanos(long d) { return x(d, C2/C0, MAX/(C2/C0)); }
- public long toMicros(long d) { return x(d, C2/C1, MAX/(C2/C1)); }
- public long toMillis(long d) { return d; }
- public long toSeconds(long d) { return d/(C3/C2); }
- public long toMinutes(long d) { return d/(C4/C2); }
- public long toHours(long d) { return d/(C5/C2); }
- public long toDays(long d) { return d/(C6/C2); }
- public long convert(long d, TimeUnit u) { return u.toMillis(d); }
- int excessNanos(long d, long m) { return 0; }
- };
- public static final TimeUnit SECONDS = new TimeUnit(3, "SECONDS") {
- private final static long serialVersionUID = 227755028449378390L;
- public long toNanos(long d) { return x(d, C3/C0, MAX/(C3/C0)); }
- public long toMicros(long d) { return x(d, C3/C1, MAX/(C3/C1)); }
- public long toMillis(long d) { return x(d, C3/C2, MAX/(C3/C2)); }
- public long toSeconds(long d) { return d; }
- public long toMinutes(long d) { return d/(C4/C3); }
- public long toHours(long d) { return d/(C5/C3); }
- public long toDays(long d) { return d/(C6/C3); }
- public long convert(long d, TimeUnit u) { return u.toSeconds(d); }
- int excessNanos(long d, long m) { return 0; }
- };
- public static final TimeUnit MINUTES = new TimeUnit(4, "MINUTES") {
- private final static long serialVersionUID = 1827351566402609187L;
- public long toNanos(long d) { return x(d, C4/C0, MAX/(C4/C0)); }
- public long toMicros(long d) { return x(d, C4/C1, MAX/(C4/C1)); }
- public long toMillis(long d) { return x(d, C4/C2, MAX/(C4/C2)); }
- public long toSeconds(long d) { return x(d, C4/C3, MAX/(C4/C3)); }
- public long toMinutes(long d) { return d; }
- public long toHours(long d) { return d/(C5/C4); }
- public long toDays(long d) { return d/(C6/C4); }
- public long convert(long d, TimeUnit u) { return u.toMinutes(d); }
- int excessNanos(long d, long m) { return 0; }
- };
- public static final TimeUnit HOURS = new TimeUnit(5, "HOURS") {
- private final static long serialVersionUID = -6438436134732089810L;
- public long toNanos(long d) { return x(d, C5/C0, MAX/(C5/C0)); }
- public long toMicros(long d) { return x(d, C5/C1, MAX/(C5/C1)); }
- public long toMillis(long d) { return x(d, C5/C2, MAX/(C5/C2)); }
- public long toSeconds(long d) { return x(d, C5/C3, MAX/(C5/C3)); }
- public long toMinutes(long d) { return x(d, C5/C4, MAX/(C5/C4)); }
- public long toHours(long d) { return d; }
- public long toDays(long d) { return d/(C6/C5); }
- public long convert(long d, TimeUnit u) { return u.toHours(d); }
- int excessNanos(long d, long m) { return 0; }
- };
- public static final TimeUnit DAYS = new TimeUnit(6, "DAYS") {
- private final static long serialVersionUID = 567463171959674600L;
- public long toNanos(long d) { return x(d, C6/C0, MAX/(C6/C0)); }
- public long toMicros(long d) { return x(d, C6/C1, MAX/(C6/C1)); }
- public long toMillis(long d) { return x(d, C6/C2, MAX/(C6/C2)); }
- public long toSeconds(long d) { return x(d, C6/C3, MAX/(C6/C3)); }
- public long toMinutes(long d) { return x(d, C6/C4, MAX/(C6/C4)); }
- public long toHours(long d) { return x(d, C6/C5, MAX/(C6/C5)); }
- public long toDays(long d) { return d; }
- public long convert(long d, TimeUnit u) { return u.toDays(d); }
- int excessNanos(long d, long m) { return 0; }
- };
-
- private static final TimeUnit[] values = new TimeUnit[]
- { NANOSECONDS, MICROSECONDS, MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS };
-
- public static TimeUnit[] values() {
- return (TimeUnit[])values.clone();
- }
-
- /**
- * Returns the enum constant of this type with the specified name. The
- * string must match <em>exactly</em> an identifier used to declare an
- * enum constant in this type. (Extraneous whitespace characters are not
- * permitted.)
- *
- * @param name the name of the enum constant to be returned
- * @return the enum constant with the specified name
- * @throws IllegalArgumentException
- * if this enum type has no constant with the specified name
- */
- public static TimeUnit valueOf(String name) {
- for (int i = 0; i < values.length; i++) {
- if (values[i].name.equals(name)) {
- return values[i];
- }
- }
- throw new IllegalArgumentException("No enum const TimeUnit." + name);
- }
-
- /**
- * The ordinal of this unit. This is useful both for {@link #ordinal()}
- * and to maintain serialization consistence with earlier versions.
- */
- private final int index;
-
- /** name of this unit */
- private final String name;
-
- /** Internal constructor */
- TimeUnit(int index, String name) {
- this.index = index;
- this.name = name;
- }
-
- // Handy constants for conversion methods
- static final long C0 = 1;
- static final long C1 = C0 * 1000;
- static final long C2 = C1 * 1000;
- static final long C3 = C2 * 1000;
- static final long C4 = C3 * 60;
- static final long C5 = C4 * 60;
- static final long C6 = C5 * 24;
-
- static final long MAX = Long.MAX_VALUE;
-
- /**
- * Scale d by m, checking for overflow.
- * This has a short name to make above code more readable.
- */
- static long x(long d, long m, long over) {
- if (d > over) return Long.MAX_VALUE;
- if (d < -over) return Long.MIN_VALUE;
- return d * m;
- }
-
- /**
- * Convert the given time duration in the given unit to this
- * unit. Conversions from finer to coarser granularities
- * truncate, so lose precision. For example converting
- * <tt>999</tt> milliseconds to seconds results in
- * <tt>0</tt>. Conversions from coarser to finer granularities
- * with arguments that would numerically overflow saturate to
- * <tt>Long.MIN_VALUE</tt> if negative or <tt>Long.MAX_VALUE</tt>
- * if positive.
- *
- * <p>For example, to convert 10 minutes to milliseconds, use:
- * <tt>TimeUnit.MILLISECONDS.convert(10L, TimeUnit.MINUTES)</tt>
- *
- * @param sourceDuration the time duration in the given <tt>sourceUnit</tt>
- * @param sourceUnit the unit of the <tt>sourceDuration</tt> argument
- * @return the converted duration in this unit,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- */
- public abstract long convert(long sourceDuration, TimeUnit sourceUnit);
-
- /**
- * Equivalent to <tt>NANOSECONDS.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- * @see #convert
- */
- public abstract long toNanos(long duration);
-
- /**
- * Equivalent to <tt>MICROSECONDS.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- * @see #convert
- */
- public abstract long toMicros(long duration);
-
- /**
- * Equivalent to <tt>MILLISECONDS.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- * @see #convert
- */
- public abstract long toMillis(long duration);
-
- /**
- * Equivalent to <tt>SECONDS.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- * @see #convert
- */
- public abstract long toSeconds(long duration);
-
- /**
- * Equivalent to <tt>MINUTES.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- * @see #convert
- * @since 1.6
- */
- public abstract long toMinutes(long duration);
-
- /**
- * Equivalent to <tt>HOURS.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration,
- * or <tt>Long.MIN_VALUE</tt> if conversion would negatively
- * overflow, or <tt>Long.MAX_VALUE</tt> if it would positively overflow.
- * @see #convert
- * @since 1.6
- */
- public abstract long toHours(long duration);
-
- /**
- * Equivalent to <tt>DAYS.convert(duration, this)</tt>.
- * @param duration the duration
- * @return the converted duration
- * @see #convert
- * @since 1.6
- */
- public abstract long toDays(long duration);
-
- /**
- * Utility to compute the excess-nanosecond argument to wait,
- * sleep, join.
- * @param d the duration
- * @param m the number of milliseconds
- * @return the number of nanoseconds
- */
- abstract int excessNanos(long d, long m);
-
- /**
- * Returns the name of this enum constant, exactly as declared in its enum
- * declaration. <strong>Most programmers should use the
- * {@link #toString()} method in preference to this one, as the toString
- * method may return a more user-friendly name.</strong> This method is
- * designed primarily for use in specialized situations where correctness
- * depends on getting the exact name, which will not vary from release to
- * release.
- *
- * @return the name of this enum constant
- */
- public String name() {
- return name;
- }
-
- /**
- * Returns the ordinal of this enumeration constant (its position in its
- * enum declaration, where the initial constant is assigned an ordinal of
- * zero). Most programmers will have no use for this method. It is
- * designed for use by sophisticated enum-based data structures, such as
- * <code>EnumSet</code> and <code>EnumMap</code>.
- *
- * @return the ordinal of this enumeration constant
- */
- public int ordinal() {
- return index;
- }
-
- /*
- * Guarantees that deserialized objects will be referentially equal to the
- * standard enumeration objects.
- */
- protected Object readResolve() throws ObjectStreamException {
- try {
- return valueOf(name);
- } catch (IllegalArgumentException e) {
- throw new InvalidObjectException(name
- + " is not a valid enum for TimeUnit");
- }
- }
-
- /**
- * Performs a timed <tt>Object.wait</tt> using this time unit.
- * This is a convenience method that converts timeout arguments
- * into the form required by the <tt>Object.wait</tt> method.
- *
- * <p>For example, you could implement a blocking <tt>poll</tt>
- * method (see {@link BlockingQueue#poll BlockingQueue.poll})
- * using:
- *
- * <pre> public synchronized Object poll(long timeout, TimeUnit unit) throws InterruptedException {
- * while (empty) {
- * unit.timedWait(this, timeout);
- * ...
- * }
- * }</pre>
- *
- * @param obj the object to wait on
- * @param timeout the maximum time to wait. If less than
- * or equal to zero, do not wait at all.
- * @throws InterruptedException if interrupted while waiting.
- * @see java.lang.Object#wait(long, int)
- */
- public void timedWait(Object obj, long timeout)
- throws InterruptedException {
- if (timeout > 0) {
- long ms = toMillis(timeout);
- int ns = excessNanos(timeout, ms);
- obj.wait(ms, ns);
- }
- }
-
- /**
- * Performs a timed <tt>Thread.join</tt> using this time unit.
- * This is a convenience method that converts time arguments into the
- * form required by the <tt>Thread.join</tt> method.
- * @param thread the thread to wait for
- * @param timeout the maximum time to wait. If less than
- * or equal to zero, do not wait at all.
- * @throws InterruptedException if interrupted while waiting.
- * @see java.lang.Thread#join(long, int)
- */
- public void timedJoin(Thread thread, long timeout)
- throws InterruptedException {
- if (timeout > 0) {
- long ms = toMillis(timeout);
- int ns = excessNanos(timeout, ms);
- thread.join(ms, ns);
- }
- }
-
- /**
- * Performs a <tt>Thread.sleep</tt> using this unit.
- * This is a convenience method that converts time arguments into the
- * form required by the <tt>Thread.sleep</tt> method.
- * @param timeout the maximum time to sleep. If less than