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;

/**
 * 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.
     */

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

    /**
     * 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.nextWorkerName());
        setDaemon(true);
        Thread.UncaughtExceptionHandler ueh = pool.ueh;
        if (ueh != null)
            setUncaughtExceptionHandler(ueh);
        this.pool = pool;
        pool.registerWorker(this.workQueue = new ForkJoinPool.WorkQueue
                            (pool, this, pool.localMode));
    }

    /**
     * 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() {
        return workQueue.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 {
            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.1
Committed:	Mon Aug 13 15:52:33 2012 UTC (11 years, 8 months ago) by dl
Branch:	MAIN
Log Message:	Introduce CHM bulk parallel APIs
#	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			/**
10			* A thread managed by a {@link ForkJoinPool}, which executes
11			* {@link ForkJoinTask}s.
12			* This class is subclassable solely for the sake of adding
13			* functionality -- there are no overridable methods dealing with
14			* scheduling or execution. However, you can override initialization
15			* and termination methods surrounding the main task processing loop.
16			* If you do create such a subclass, you will also need to supply a
17			* custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to use it
18			* in a {@code ForkJoinPool}.
19			*
20			* @since 1.7
21			* @author Doug Lea
22			*/
23			public class ForkJoinWorkerThread extends Thread {
24			/*
25			* ForkJoinWorkerThreads are managed by ForkJoinPools and perform
26			* ForkJoinTasks. For explanation, see the internal documentation
27			* of class ForkJoinPool.
28			*/
29
30			final ForkJoinPool.WorkQueue workQueue; // Work-stealing mechanics
31			final ForkJoinPool pool; // the pool this thread works in
32
33			/**
34			* Creates a ForkJoinWorkerThread operating in the given pool.
35			*
36			* @param pool the pool this thread works in
37			* @throws NullPointerException if pool is null
38			*/
39			protected ForkJoinWorkerThread(ForkJoinPool pool) {
40			super(pool.nextWorkerName());
41			setDaemon(true);
42			Thread.UncaughtExceptionHandler ueh = pool.ueh;
43			if (ueh != null)
44			setUncaughtExceptionHandler(ueh);
45			this.pool = pool;
46			pool.registerWorker(this.workQueue = new ForkJoinPool.WorkQueue
47			(pool, this, pool.localMode));
48			}
49
50			/**
51			* Returns the pool hosting this thread.
52			*
53			* @return the pool
54			*/
55			public ForkJoinPool getPool() {
56			return pool;
57			}
58
59			/**
60			* Returns the index number of this thread in its pool. The
61			* returned value ranges from zero to the maximum number of
62			* threads (minus one) that have ever been created in the pool.
63			* This method may be useful for applications that track status or
64			* collect results per-worker rather than per-task.
65			*
66			* @return the index number
67			*/
68			public int getPoolIndex() {
69			return workQueue.poolIndex;
70			}
71
72			/**
73			* Initializes internal state after construction but before
74			* processing any tasks. If you override this method, you must
75			* invoke {@code super.onStart()} at the beginning of the method.
76			* Initialization requires care: Most fields must have legal
77			* default values, to ensure that attempted accesses from other
78			* threads work correctly even before this thread starts
79			* processing tasks.
80			*/
81			protected void onStart() {
82			}
83
84			/**
85			* Performs cleanup associated with termination of this worker
86			* thread. If you override this method, you must invoke
87			* {@code super.onTermination} at the end of the overridden method.
88			*
89			* @param exception the exception causing this thread to abort due
90			* to an unrecoverable error, or {@code null} if completed normally
91			*/
92			protected void onTermination(Throwable exception) {
93			}
94
95			/**
96			* This method is required to be public, but should never be
97			* called explicitly. It performs the main run loop to execute
98			* {@link ForkJoinTask}s.
99			*/
100			public void run() {
101			Throwable exception = null;
102			try {
103			onStart();
104			pool.runWorker(workQueue);
105			} catch (Throwable ex) {
106			exception = ex;
107			} finally {
108			try {
109			onTermination(exception);
110			} catch (Throwable ex) {
111			if (exception == null)
112			exception = ex;
113			} finally {
114			pool.deregisterWorker(this, exception);
115			}
116			}
117			}
118			}