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.AccessController;
import java.security.AccessControlContext;
import java.security.PrivilegedAction;
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(int, ForkJoinWorkerThreadFactory,
 * UncaughtExceptionHandler, boolean, int, int, int, Predicate, long, TimeUnit)
 * 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.
     */

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

    /**
     * Full nonpublic constructor.
     */
    ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
                         boolean useSystemClassLoader, boolean isInnocuous) {
        super(group, null, pool.nextWorkerThreadName(), 0L, !isInnocuous);
        UncaughtExceptionHandler handler = (this.pool = pool).ueh;
        this.workQueue = new ForkJoinPool.WorkQueue(this, 0);
        super.setDaemon(true);
        if (handler != null)
            super.setUncaughtExceptionHandler(handler);
        if (useSystemClassLoader)
            super.setContextClassLoader(ClassLoader.getSystemClassLoader());
    }

    /**
     * Creates a ForkJoinWorkerThread operating in the given thread group and
     * pool, and with the given policy for preserving ThreadLocals.
     *
     * @param group if non-null, the thread group for this thread
     * @param pool the pool this thread works in
     * @param preserveThreadLocals if true, always preserve the values of
     * ThreadLocal variables across tasks; otherwise they may be cleared.
     * @throws NullPointerException if pool is null
     * @since 19
     */
    protected ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
                                   boolean preserveThreadLocals) {
        this(group, pool, false, !preserveThreadLocals);
    }

    /**
     * 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) {
        this(null, pool, false, false);
    }

    /**
     * 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() {
        Throwable exception = null;
        ForkJoinPool p = pool;
        ForkJoinPool.WorkQueue w = workQueue;
        if (p != null && w != null) {   // skip on failed initialization
            try {
                p.registerWorker(w);
                onStart();
                p.runWorker(w);
            } catch (Throwable ex) {
                exception = ex;
            } finally {
                try {
                    onTermination(exception);
                } catch (Throwable ex) {
                    if (exception == null)
                        exception = ex;
                } finally {
                    p.deregisterWorker(this, exception);
                }
            }
        }
    }

    /**
     * A worker thread that has no permissions, is not a member of any
     * user-defined ThreadGroup, uses the system class loader as
     * thread context class loader, 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;
        @SuppressWarnings("removal")
        private static final AccessControlContext innocuousACC;
        InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
            super(innocuousThreadGroup, pool, true, true);
        }

        @Override @SuppressWarnings("removal")
        protected void onStart() {
            ForkJoinPool.WorkQueue w = workQueue;
            if (w != null)
                w.setInnocuous();
            Thread t = Thread.currentThread();
            ThreadLocalRandom.setInheritedAccessControlContext(t, innocuousACC);
        }

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

        @Override // paranoically
        public void setContextClassLoader(ClassLoader cl) {
            if (cl != null && ClassLoader.getSystemClassLoader() != cl)
                throw new SecurityException("setContextClassLoader");
        }

        @SuppressWarnings("removal")
        static AccessControlContext createACC() {
            return new AccessControlContext(
                new ProtectionDomain[] { new ProtectionDomain(null, null) });
        }
        static ThreadGroup createGroup() {
            ThreadGroup group = Thread.currentThread().getThreadGroup();
            for (ThreadGroup p; (p = group.getParent()) != null; )
                group = p;
            return new ThreadGroup(group, "InnocuousForkJoinWorkerThreadGroup");
        }
        static {
            @SuppressWarnings("removal")
            SecurityManager sm = System.getSecurityManager();
            @SuppressWarnings("removal")
            ThreadGroup g = innocuousThreadGroup =
                (sm == null) ? createGroup() :
                AccessController.doPrivileged(new PrivilegedAction<>() {
                        public ThreadGroup run() {
                            return createGroup(); }});
            @SuppressWarnings("removal")
            AccessControlContext a = innocuousACC =
                (sm == null) ? createACC() :
                AccessController.doPrivileged(new PrivilegedAction<>() {
                        public AccessControlContext run() {
                            return createACC(); }});
        }
    }
}
Revision:	1.84
Committed:	Fri Apr 29 17:00:36 2022 UTC (2 years, 1 month ago) by dl
Branch:	MAIN
Changes since 1.83:	+11 -8 lines
Log Message:	Misc touchups
#	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.AccessController;
10	import java.security.AccessControlContext;
11	import java.security.PrivilegedAction;
12	import java.security.ProtectionDomain;
13
14	/**
15	* A thread managed by a {@link ForkJoinPool}, which executes
16	* {@link ForkJoinTask}s.
17	* This class is subclassable solely for the sake of adding
18	* functionality -- there are no overridable methods dealing with
19	* scheduling or execution. However, you can override initialization
20	* and termination methods surrounding the main task processing loop.
21	* If you do create such a subclass, you will also need to supply a
22	* custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to
23	* {@linkplain ForkJoinPool#ForkJoinPool(int, ForkJoinWorkerThreadFactory,
24	* UncaughtExceptionHandler, boolean, int, int, int, Predicate, long, TimeUnit)
25	* use it} in a {@code ForkJoinPool}.
26	*
27	* @since 1.7
28	* @author Doug Lea
29	*/
30	public class ForkJoinWorkerThread extends Thread {
31	/*
32	* ForkJoinWorkerThreads are managed by ForkJoinPools and perform
33	* ForkJoinTasks. For explanation, see the internal documentation
34	* of class ForkJoinPool.
35	*
36	* This class just maintains links to its pool and WorkQueue.
37	*/
38
39	final ForkJoinPool pool; // the pool this thread works in
40	final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
41
42	/**
43	* Full nonpublic constructor.
44	*/
45	ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
46	boolean useSystemClassLoader, boolean isInnocuous) {
47	super(group, null, pool.nextWorkerThreadName(), 0L, !isInnocuous);
48	UncaughtExceptionHandler handler = (this.pool = pool).ueh;
49	this.workQueue = new ForkJoinPool.WorkQueue(this, 0);
50	super.setDaemon(true);
51	if (handler != null)
52	super.setUncaughtExceptionHandler(handler);
53	if (useSystemClassLoader)
54	super.setContextClassLoader(ClassLoader.getSystemClassLoader());
55	}
56
57	/**
58	* Creates a ForkJoinWorkerThread operating in the given thread group and
59	* pool, and with the given policy for preserving ThreadLocals.
60	*
61	* @param group if non-null, the thread group for this thread
62	* @param pool the pool this thread works in
63	* @param preserveThreadLocals if true, always preserve the values of
64	* ThreadLocal variables across tasks; otherwise they may be cleared.
65	* @throws NullPointerException if pool is null
66	* @since 19
67	*/
68	protected ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
69	boolean preserveThreadLocals) {
70	this(group, pool, false, !preserveThreadLocals);
71	}
72
73	/**
74	* Creates a ForkJoinWorkerThread operating in the given pool.
75	*
76	* @param pool the pool this thread works in
77	* @throws NullPointerException if pool is null
78	*/
79	protected ForkJoinWorkerThread(ForkJoinPool pool) {
80	this(null, pool, false, false);
81	}
82
83	/**
84	* Returns the pool hosting this thread.
85	*
86	* @return the pool
87	*/
88	public ForkJoinPool getPool() {
89	return pool;
90	}
91
92	/**
93	* Returns the unique index number of this thread in its pool.
94	* The returned value ranges from zero to the maximum number of
95	* threads (minus one) that may exist in the pool, and does not
96	* change during the lifetime of the thread. This method may be
97	* useful for applications that track status or collect results
98	* per-worker-thread rather than per-task.
99	*
100	* @return the index number
101	*/
102	public int getPoolIndex() {
103	return workQueue.getPoolIndex();
104	}
105
106	/**
107	* Initializes internal state after construction but before
108	* processing any tasks. If you override this method, you must
109	* invoke {@code super.onStart()} at the beginning of the method.
110	* Initialization requires care: Most fields must have legal
111	* default values, to ensure that attempted accesses from other
112	* threads work correctly even before this thread starts
113	* processing tasks.
114	*/
115	protected void onStart() {
116	}
117
118	/**
119	* Performs cleanup associated with termination of this worker
120	* thread. If you override this method, you must invoke
121	* {@code super.onTermination} at the end of the overridden method.
122	*
123	* @param exception the exception causing this thread to abort due
124	* to an unrecoverable error, or {@code null} if completed normally
125	*/
126	protected void onTermination(Throwable exception) {
127	}
128
129	/**
130	* This method is required to be public, but should never be
131	* called explicitly. It performs the main run loop to execute
132	* {@link ForkJoinTask}s.
133	*/
134	public void run() {
135	Throwable exception = null;
136	ForkJoinPool p = pool;
137	ForkJoinPool.WorkQueue w = workQueue;
138	if (p != null && w != null) { // skip on failed initialization
139	try {
140	p.registerWorker(w);
141	onStart();
142	p.runWorker(w);
143	} catch (Throwable ex) {
144	exception = ex;
145	} finally {
146	try {
147	onTermination(exception);
148	} catch (Throwable ex) {
149	if (exception == null)
150	exception = ex;
151	} finally {
152	p.deregisterWorker(this, exception);
153	}
154	}
155	}
156	}
157
158	/**
159	* A worker thread that has no permissions, is not a member of any
160	* user-defined ThreadGroup, uses the system class loader as
161	* thread context class loader, and erases all ThreadLocals after
162	* running each top-level task.
163	*/
164	static final class InnocuousForkJoinWorkerThread extends ForkJoinWorkerThread {
165	/** The ThreadGroup for all InnocuousForkJoinWorkerThreads */
166	private static final ThreadGroup innocuousThreadGroup;
167	@SuppressWarnings("removal")
168	private static final AccessControlContext innocuousACC;
169	InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
170	super(innocuousThreadGroup, pool, true, true);
171	}
172
173	@Override @SuppressWarnings("removal")
174	protected void onStart() {
175	ForkJoinPool.WorkQueue w = workQueue;
176	if (w != null)
177	w.setInnocuous();
178	Thread t = Thread.currentThread();
179	ThreadLocalRandom.setInheritedAccessControlContext(t, innocuousACC);
180	}
181
182	@Override // to silently fail
183	public void setUncaughtExceptionHandler(UncaughtExceptionHandler x) { }
184
185	@Override // paranoically
186	public void setContextClassLoader(ClassLoader cl) {
187	if (cl != null && ClassLoader.getSystemClassLoader() != cl)
188	throw new SecurityException("setContextClassLoader");
189	}
190
191	@SuppressWarnings("removal")
192	static AccessControlContext createACC() {
193	return new AccessControlContext(
194	new ProtectionDomain[] { new ProtectionDomain(null, null) });
195	}
196	static ThreadGroup createGroup() {
197	ThreadGroup group = Thread.currentThread().getThreadGroup();
198	for (ThreadGroup p; (p = group.getParent()) != null; )
199	group = p;
200	return new ThreadGroup(group, "InnocuousForkJoinWorkerThreadGroup");
201	}
202	static {
203	@SuppressWarnings("removal")
204	SecurityManager sm = System.getSecurityManager();
205	@SuppressWarnings("removal")
206	ThreadGroup g = innocuousThreadGroup =
207	(sm == null) ? createGroup() :
208	AccessController.doPrivileged(new PrivilegedAction<>() {
209	public ThreadGroup run() {
210	return createGroup(); }});
211	@SuppressWarnings("removal")
212	AccessControlContext a = innocuousACC =
213	(sm == null) ? createACC() :
214	AccessController.doPrivileged(new PrivilegedAction<>() {
215	public AccessControlContext run() {
216	return createACC(); }});
217	}
218	}
219	}