util/concurrent/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 java.util.concurrent;

import java.security.AccessControlContext;
import java.security.ProtectionDomain;

/**
 * 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
 * {@linkplain ForkJoinPool#ForkJoinPool 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 immediately upon construction, but the
     * workQueue field is not set until a call to registerWorker
     * completes. This leads to a visibility race, that is tolerated
     * by requiring that the workQueue field is only accessed by the
     * owning thread.
     *
     * Support for (non-public) subclass InnocuousForkJoinWorkerThread
     * requires that we break quite a lot of encapsulation (via helper
     * methods in ThreadLocalRandom) both here and in the subclass to
     * access and set Thread fields.
     */

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

    /**
     * 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) {
        // Use a placeholder until a useful name can be set in registerWorker
        super("aForkJoinWorkerThread");
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }

    /**
     * Version for InnocuousForkJoinWorkerThread.
     */
    ForkJoinWorkerThread(ForkJoinPool pool, ThreadGroup threadGroup,
                         AccessControlContext acc) {
        super(threadGroup, null, "aForkJoinWorkerThread");
        ThreadLocalRandom.setInheritedAccessControlContext(this, acc);
        ThreadLocalRandom.eraseThreadLocals(this); // clear before registering
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }

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

    /**
     * Returns the unique index number of this thread in its pool.
     * The returned value ranges from zero to the maximum number of
     * threads (minus one) that may exist in the pool, and does not
     * change during the lifetime of the thread.  This method may be
     * useful for applications that track status or collect results
     * per-worker-thread rather than per-task.
     *
     * @return the index number
     */
    public int getPoolIndex() {
        return workQueue.getPoolIndex();
    }

    /**
     * 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() {
        if (workQueue.array == null) { // only run once
            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);
                }
            }
        }
    }

    /**
     * Non-public hook method for InnocuousForkJoinWorkerThread.
     */
    void afterTopLevelExec() {
    }

    /**
     * A worker thread that has no permissions, is not a member of any
     * user-defined ThreadGroup, and erases all ThreadLocals after
     * running each top-level task.
     */
    static final class InnocuousForkJoinWorkerThread extends ForkJoinWorkerThread {
        /** The ThreadGroup for all InnocuousForkJoinWorkerThreads */
        private static final ThreadGroup innocuousThreadGroup =
            ThreadLocalRandom.createThreadGroup("InnocuousForkJoinWorkerThreadGroup");

        /** An AccessControlContext supporting no privileges */
        private static final AccessControlContext INNOCUOUS_ACC =
            new AccessControlContext(
                new ProtectionDomain[] {
                    new ProtectionDomain(null, null)
                });

        InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
            super(pool, innocuousThreadGroup, INNOCUOUS_ACC);
        }

        @Override // to erase ThreadLocals
        void afterTopLevelExec() {
            ThreadLocalRandom.eraseThreadLocals(this);
        }

        @Override // to always report system loader
        public ClassLoader getContextClassLoader() {
            return ClassLoader.getSystemClassLoader();
        }

        @Override // to silently fail
        public void setUncaughtExceptionHandler(UncaughtExceptionHandler x) { }

        @Override // paranoically
        public void setContextClassLoader(ClassLoader cl) {
            throw new SecurityException("setContextClassLoader");
        }

    }
}
Revision:	1.72
Committed:	Fri Jun 3 17:18:52 2016 UTC (8 years ago) by jsr166
Branch:	MAIN
Changes since 1.71:	+1 -1 lines
Log Message:	comment tidy
#	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 java.util.concurrent;
8
9	import java.security.AccessControlContext;
10	import java.security.ProtectionDomain;
11
12	/**
13	* A thread managed by a {@link ForkJoinPool}, which executes
14	* {@link ForkJoinTask}s.
15	* This class is subclassable solely for the sake of adding
16	* functionality -- there are no overridable methods dealing with
17	* scheduling or execution. However, you can override initialization
18	* and termination methods surrounding the main task processing loop.
19	* If you do create such a subclass, you will also need to supply a
20	* custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to
21	* {@linkplain ForkJoinPool#ForkJoinPool use it} in a {@code ForkJoinPool}.
22	*
23	* @since 1.7
24	* @author Doug Lea
25	*/
26	public class ForkJoinWorkerThread extends Thread {
27	/*
28	* ForkJoinWorkerThreads are managed by ForkJoinPools and perform
29	* ForkJoinTasks. For explanation, see the internal documentation
30	* of class ForkJoinPool.
31	*
32	* This class just maintains links to its pool and WorkQueue. The
33	* pool field is set immediately upon construction, but the
34	* workQueue field is not set until a call to registerWorker
35	* completes. This leads to a visibility race, that is tolerated
36	* by requiring that the workQueue field is only accessed by the
37	* owning thread.
38	*
39	* Support for (non-public) subclass InnocuousForkJoinWorkerThread
40	* requires that we break quite a lot of encapsulation (via helper
41	* methods in ThreadLocalRandom) both here and in the subclass to
42	* access and set Thread fields.
43	*/
44
45	final ForkJoinPool pool; // the pool this thread works in
46	final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
47
48	/**
49	* 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	// Use a placeholder until a useful name can be set in registerWorker
56	super("aForkJoinWorkerThread");
57	this.pool = pool;
58	this.workQueue = pool.registerWorker(this);
59	}
60
61	/**
62	* Version for InnocuousForkJoinWorkerThread.
63	*/
64	ForkJoinWorkerThread(ForkJoinPool pool, ThreadGroup threadGroup,
65	AccessControlContext acc) {
66	super(threadGroup, null, "aForkJoinWorkerThread");
67	ThreadLocalRandom.setInheritedAccessControlContext(this, acc);
68	ThreadLocalRandom.eraseThreadLocals(this); // clear before registering
69	this.pool = pool;
70	this.workQueue = pool.registerWorker(this);
71	}
72
73	/**
74	* Returns the pool hosting this thread.
75	*
76	* @return the pool
77	*/
78	public ForkJoinPool getPool() {
79	return pool;
80	}
81
82	/**
83	* Returns the unique index number of this thread in its pool.
84	* The returned value ranges from zero to the maximum number of
85	* threads (minus one) that may exist in the pool, and does not
86	* change during the lifetime of the thread. This method may be
87	* useful for applications that track status or collect results
88	* per-worker-thread rather than per-task.
89	*
90	* @return the index number
91	*/
92	public int getPoolIndex() {
93	return workQueue.getPoolIndex();
94	}
95
96	/**
97	* Initializes internal state after construction but before
98	* processing any tasks. If you override this method, you must
99	* invoke {@code super.onStart()} at the beginning of the method.
100	* Initialization requires care: Most fields must have legal
101	* default values, to ensure that attempted accesses from other
102	* threads work correctly even before this thread starts
103	* processing tasks.
104	*/
105	protected void onStart() {
106	}
107
108	/**
109	* Performs cleanup associated with termination of this worker
110	* thread. If you override this method, you must invoke
111	* {@code super.onTermination} at the end of the overridden method.
112	*
113	* @param exception the exception causing this thread to abort due
114	* to an unrecoverable error, or {@code null} if completed normally
115	*/
116	protected void onTermination(Throwable exception) {
117	}
118
119	/**
120	* This method is required to be public, but should never be
121	* called explicitly. It performs the main run loop to execute
122	* {@link ForkJoinTask}s.
123	*/
124	public void run() {
125	if (workQueue.array == null) { // only run once
126	Throwable exception = null;
127	try {
128	onStart();
129	pool.runWorker(workQueue);
130	} catch (Throwable ex) {
131	exception = ex;
132	} finally {
133	try {
134	onTermination(exception);
135	} catch (Throwable ex) {
136	if (exception == null)
137	exception = ex;
138	} finally {
139	pool.deregisterWorker(this, exception);
140	}
141	}
142	}
143	}
144
145	/**
146	* Non-public hook method for InnocuousForkJoinWorkerThread.
147	*/
148	void afterTopLevelExec() {
149	}
150
151	/**
152	* A worker thread that has no permissions, is not a member of any
153	* user-defined ThreadGroup, and erases all ThreadLocals after
154	* running each top-level task.
155	*/
156	static final class InnocuousForkJoinWorkerThread extends ForkJoinWorkerThread {
157	/** The ThreadGroup for all InnocuousForkJoinWorkerThreads */
158	private static final ThreadGroup innocuousThreadGroup =
159	ThreadLocalRandom.createThreadGroup("InnocuousForkJoinWorkerThreadGroup");
160
161	/** An AccessControlContext supporting no privileges */
162	private static final AccessControlContext INNOCUOUS_ACC =
163	new AccessControlContext(
164	new ProtectionDomain[] {
165	new ProtectionDomain(null, null)
166	});
167
168	InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
169	super(pool, innocuousThreadGroup, INNOCUOUS_ACC);
170	}
171
172	@Override // to erase ThreadLocals
173	void afterTopLevelExec() {
174	ThreadLocalRandom.eraseThreadLocals(this);
175	}
176
177	@Override // to always report system loader
178	public ClassLoader getContextClassLoader() {
179	return ClassLoader.getSystemClassLoader();
180	}
181
182	@Override // to silently fail
183	public void setUncaughtExceptionHandler(UncaughtExceptionHandler x) { }
184
185	@Override // paranoically
186	public void setContextClassLoader(ClassLoader cl) {
187	throw new SecurityException("setContextClassLoader");
188	}
189
190	}
191	}