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

    /**
     * An initial name for a newly constructed worker, used until
     * onStart can establish a useful name. This removes need to
     * establish a name from worker startup path.
     */
    static final String provisionalName = "aForkJoinWorkerThread";

    /**
     * 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(provisionalName); // bootstrap name
        Thread.UncaughtExceptionHandler ueh = pool.ueh;
        if (ueh != null)
            setUncaughtExceptionHandler(ueh);
        setDaemon(true);
        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() {
        String pref; // replace bootstrap name
        if (provisionalName.equals(getName()) &&
            (pref = pool.workerNamePrefix) != null)
            setName(pref.concat(Long.toString(getId())));
    }

    /**
     * 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.2
Committed:	Wed Nov 14 17:20:29 2012 UTC (11 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.1:	+13 -2 lines
Log Message:	commonPool support
#	Content
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	* An initial name for a newly constructed worker, used until
35	* onStart can establish a useful name. This removes need to
36	* establish a name from worker startup path.
37	*/
38	static final String provisionalName = "aForkJoinWorkerThread";
39
40	/**
41	* Creates a ForkJoinWorkerThread operating in the given pool.
42	*
43	* @param pool the pool this thread works in
44	* @throws NullPointerException if pool is null
45	*/
46	protected ForkJoinWorkerThread(ForkJoinPool pool) {
47	super(provisionalName); // bootstrap name
48	Thread.UncaughtExceptionHandler ueh = pool.ueh;
49	if (ueh != null)
50	setUncaughtExceptionHandler(ueh);
51	setDaemon(true);
52	this.pool = pool;
53	pool.registerWorker(this.workQueue = new ForkJoinPool.WorkQueue
54	(pool, this, pool.localMode));
55	}
56
57	/**
58	* Returns the pool hosting this thread.
59	*
60	* @return the pool
61	*/
62	public ForkJoinPool getPool() {
63	return pool;
64	}
65
66	/**
67	* Returns the index number of this thread in its pool. The
68	* returned value ranges from zero to the maximum number of
69	* threads (minus one) that have ever been created in the pool.
70	* This method may be useful for applications that track status or
71	* collect results per-worker rather than per-task.
72	*
73	* @return the index number
74	*/
75	public int getPoolIndex() {
76	return workQueue.poolIndex;
77	}
78
79	/**
80	* Initializes internal state after construction but before
81	* processing any tasks. If you override this method, you must
82	* invoke {@code super.onStart()} at the beginning of the method.
83	* Initialization requires care: Most fields must have legal
84	* default values, to ensure that attempted accesses from other
85	* threads work correctly even before this thread starts
86	* processing tasks.
87	*/
88	protected void onStart() {
89	String pref; // replace bootstrap name
90	if (provisionalName.equals(getName()) &&
91	(pref = pool.workerNamePrefix) != null)
92	setName(pref.concat(Long.toString(getId())));
93	}
94
95	/**
96	* Performs cleanup associated with termination of this worker
97	* thread. If you override this method, you must invoke
98	* {@code super.onTermination} at the end of the overridden method.
99	*
100	* @param exception the exception causing this thread to abort due
101	* to an unrecoverable error, or {@code null} if completed normally
102	*/
103	protected void onTermination(Throwable exception) {
104	}
105
106	/**
107	* This method is required to be public, but should never be
108	* called explicitly. It performs the main run loop to execute
109	* {@link ForkJoinTask}s.
110	*/
111	public void run() {
112	Throwable exception = null;
113	try {
114	onStart();
115	pool.runWorker(workQueue);
116	} catch (Throwable ex) {
117	exception = ex;
118	} finally {
119	try {
120	onTermination(exception);
121	} catch (Throwable ex) {
122	if (exception == null)
123	exception = ex;
124	} finally {
125	pool.deregisterWorker(this, exception);
126	}
127	}
128	}
129	}