src/jsr166e/ForkJoinWorkerThread.java

/*
 * 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/publicdomain/zero/1.0/
 */

package jsr166e;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * A thread managed by a {@link ForkJoinPool}, which executes
 * {@link ForkJoinTask}s.
 * This class is subclassable solely for the sake of adding
 * functionality -- there are no overridable methods dealing with
 * scheduling or execution.  However, you can override initialization
 * and termination methods surrounding the main task processing loop.
 * If you do create such a subclass, you will also need to supply a
 * custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to use it
 * in a {@code ForkJoinPool}.
 *
 * @since 1.7
 * @author Doug Lea
 */
public class ForkJoinWorkerThread extends Thread {
    /*
     * ForkJoinWorkerThreads are managed by ForkJoinPools and perform
     * ForkJoinTasks. For explanation, see the internal documentation
     * of class ForkJoinPool.
     *
     * This class just maintains links to its pool and WorkQueue.  The
     * pool field is set upon construction, but the workQueue field is
     * not set until the thread has started (unless forced early by a
     * subclass constructor call to poolIndex()).  This provides
     * better memory placement (because this thread allocates queue
     * and bookkeeping fields) but because the field is non-final, we
     * require that it never be accessed except by the owning thread.
     */

    final ForkJoinPool pool;                // the pool this thread works in
    ForkJoinPool.WorkQueue workQueue;       // Work-stealing mechanics

    /**
     * Sequence number for creating worker Names
     */
    private static final AtomicInteger threadNumber = new AtomicInteger();

    /**
     * Creates a ForkJoinWorkerThread operating in the given pool.
     *
     * @param pool the pool this thread works in
     * @throws NullPointerException if pool is null
     */
    protected ForkJoinWorkerThread(ForkJoinPool pool) {
        super(pool.workerNamePrefix.concat(Integer.toString(threadNumber.incrementAndGet())));
        setDaemon(true);
        this.pool = pool;
        Thread.UncaughtExceptionHandler ueh = pool.ueh;
        if (ueh != null)
            setUncaughtExceptionHandler(ueh);
    }

    /**
     * Returns the pool hosting this thread.
     *
     * @return the pool
     */
    public ForkJoinPool getPool() {
        return pool;
    }

    /**
     * Returns the index number of this thread in its pool.  The
     * returned value ranges from zero to the maximum number of
     * threads (minus one) that have ever been created in the pool.
     * This method may be useful for applications that track status or
     * collect results per-worker rather than per-task.
     *
     * @return the index number
     */
    public int getPoolIndex() {
        // force early registration if called before started
        ForkJoinPool.WorkQueue q;
        if ((q = workQueue) == null) {
            pool.registerWorker(this);
            q = workQueue;
        }
        return q.poolIndex;
    }

    /**
     * Initializes internal state after construction but before
     * processing any tasks. If you override this method, you must
     * invoke {@code super.onStart()} at the beginning of the method.
     * Initialization requires care: Most fields must have legal
     * default values, to ensure that attempted accesses from other
     * threads work correctly even before this thread starts
     * processing tasks.
     */
    protected void onStart() {
    }

    /**
     * Performs cleanup associated with termination of this worker
     * thread.  If you override this method, you must invoke
     * {@code super.onTermination} at the end of the overridden method.
     *
     * @param exception the exception causing this thread to abort due
     * to an unrecoverable error, or {@code null} if completed normally
     */
    protected void onTermination(Throwable exception) {
    }

    /**
     * This method is required to be public, but should never be
     * called explicitly. It performs the main run loop to execute
     * {@link ForkJoinTask}s.
     */
    public void run() {
        Throwable exception = null;
        try {
            pool.registerWorker(this);
            onStart();
            pool.runWorker(workQueue);
        } catch (Throwable ex) {
            exception = ex;
        } finally {
            try {
                onTermination(exception);
            } catch (Throwable ex) {
                if (exception == null)
                    exception = ex;
            } finally {
                pool.deregisterWorker(this, exception);
            }
        }
    }
}
Revision:	1.3
Committed:	Mon Nov 19 18:12:28 2012 UTC (11 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.2:	+24 -15 lines
Log Message:	better memory positioning
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain, as explained at
4			* http://creativecommons.org/publicdomain/zero/1.0/
5			*/
6
7			package jsr166e;
8
9	dl	1.3	import java.util.concurrent.atomic.AtomicInteger;
10
11	dl	1.1	/**
12			* A thread managed by a {@link ForkJoinPool}, which executes
13			* {@link ForkJoinTask}s.
14			* This class is subclassable solely for the sake of adding
15			* functionality -- there are no overridable methods dealing with
16			* scheduling or execution. However, you can override initialization
17			* and termination methods surrounding the main task processing loop.
18			* If you do create such a subclass, you will also need to supply a
19			* custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to use it
20			* in a {@code ForkJoinPool}.
21			*
22			* @since 1.7
23			* @author Doug Lea
24			*/
25			public class ForkJoinWorkerThread extends Thread {
26			/*
27			* ForkJoinWorkerThreads are managed by ForkJoinPools and perform
28			* ForkJoinTasks. For explanation, see the internal documentation
29			* of class ForkJoinPool.
30	dl	1.3	*
31			* This class just maintains links to its pool and WorkQueue. The
32			* pool field is set upon construction, but the workQueue field is
33			* not set until the thread has started (unless forced early by a
34			* subclass constructor call to poolIndex()). This provides
35			* better memory placement (because this thread allocates queue
36			* and bookkeeping fields) but because the field is non-final, we
37			* require that it never be accessed except by the owning thread.
38	dl	1.1	*/
39
40			final ForkJoinPool pool; // the pool this thread works in
41	dl	1.3	ForkJoinPool.WorkQueue workQueue; // Work-stealing mechanics
42	dl	1.1
43			/**
44	dl	1.3	* Sequence number for creating worker Names
45	dl	1.2	*/
46	dl	1.3	private static final AtomicInteger threadNumber = new AtomicInteger();
47	dl	1.2
48			/**
49	dl	1.1	* Creates a ForkJoinWorkerThread operating in the given pool.
50			*
51			* @param pool the pool this thread works in
52			* @throws NullPointerException if pool is null
53			*/
54			protected ForkJoinWorkerThread(ForkJoinPool pool) {
55	dl	1.3	super(pool.workerNamePrefix.concat(Integer.toString(threadNumber.incrementAndGet())));
56			setDaemon(true);
57			this.pool = pool;
58	dl	1.1	Thread.UncaughtExceptionHandler ueh = pool.ueh;
59			if (ueh != null)
60			setUncaughtExceptionHandler(ueh);
61			}
62
63			/**
64			* Returns the pool hosting this thread.
65			*
66			* @return the pool
67			*/
68			public ForkJoinPool getPool() {
69			return pool;
70			}
71
72			/**
73			* Returns the index number of this thread in its pool. The
74			* returned value ranges from zero to the maximum number of
75			* threads (minus one) that have ever been created in the pool.
76			* This method may be useful for applications that track status or
77			* collect results per-worker rather than per-task.
78			*
79			* @return the index number
80			*/
81			public int getPoolIndex() {
82	dl	1.3	// force early registration if called before started
83			ForkJoinPool.WorkQueue q;
84			if ((q = workQueue) == null) {
85			pool.registerWorker(this);
86			q = workQueue;
87			}
88			return q.poolIndex;
89	dl	1.1	}
90
91			/**
92			* Initializes internal state after construction but before
93			* processing any tasks. If you override this method, you must
94			* invoke {@code super.onStart()} at the beginning of the method.
95			* Initialization requires care: Most fields must have legal
96			* default values, to ensure that attempted accesses from other
97			* threads work correctly even before this thread starts
98			* processing tasks.
99			*/
100			protected void onStart() {
101			}
102
103			/**
104			* Performs cleanup associated with termination of this worker
105			* thread. If you override this method, you must invoke
106			* {@code super.onTermination} at the end of the overridden method.
107			*
108			* @param exception the exception causing this thread to abort due
109			* to an unrecoverable error, or {@code null} if completed normally
110			*/
111			protected void onTermination(Throwable exception) {
112			}
113
114			/**
115			* This method is required to be public, but should never be
116			* called explicitly. It performs the main run loop to execute
117			* {@link ForkJoinTask}s.
118			*/
119			public void run() {
120			Throwable exception = null;
121			try {
122	dl	1.3	pool.registerWorker(this);
123	dl	1.1	onStart();
124			pool.runWorker(workQueue);
125			} catch (Throwable ex) {
126			exception = ex;
127			} finally {
128			try {
129			onTermination(exception);
130			} catch (Throwable ex) {
131			if (exception == null)
132			exception = ex;
133			} finally {
134			pool.deregisterWorker(this, exception);
135			}
136			}
137			}
138			}