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.PrivilegedAction;

/**
 * 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);
        UncaughtExceptionHandler handler = (this.pool = pool).ueh;
        this.workQueue = new ForkJoinPool.WorkQueue(this, isInnocuous);
        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.
     *
     * @param group if non-null, the thread group for this thread
     * @param pool the pool this thread works in
     * @throws NullPointerException if pool is null
     */
    /* TODO: protected */ ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool) {
        this(group, pool, false, false);
    }

    /**
     * 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 =
            AccessController.doPrivileged(new PrivilegedAction<>() {
                public ThreadGroup run() {
                    ThreadGroup group = Thread.currentThread().getThreadGroup();
                    for (ThreadGroup p; (p = group.getParent()) != null; )
                        group = p;
                    return new ThreadGroup(
                        group, "InnocuousForkJoinWorkerThreadGroup");
                }});

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

        @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");
        }
    }
}
Revision:	1.82
Committed:	Sun Dec 6 18:10:00 2020 UTC (3 years, 5 months ago) by jsr166
Branch:	MAIN
Changes since 1.81:	+1 -1 lines
Log Message:	ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool): demote to package-private, for now
#	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.PrivilegedAction;
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(int, ForkJoinWorkerThreadFactory,
22	* UncaughtExceptionHandler, boolean, int, int, int, Predicate, long, TimeUnit)
23	* use it} in a {@code ForkJoinPool}.
24	*
25	* @since 1.7
26	* @author Doug Lea
27	*/
28	public class ForkJoinWorkerThread extends Thread {
29	/*
30	* ForkJoinWorkerThreads are managed by ForkJoinPools and perform
31	* ForkJoinTasks. For explanation, see the internal documentation
32	* of class ForkJoinPool.
33	*
34	* This class just maintains links to its pool and WorkQueue.
35	*/
36
37	final ForkJoinPool pool; // the pool this thread works in
38	final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
39
40	/**
41	* Full nonpublic constructor.
42	*/
43	ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
44	boolean useSystemClassLoader, boolean isInnocuous) {
45	super(group, null, pool.nextWorkerThreadName(), 0L);
46	UncaughtExceptionHandler handler = (this.pool = pool).ueh;
47	this.workQueue = new ForkJoinPool.WorkQueue(this, isInnocuous);
48	super.setDaemon(true);
49	if (handler != null)
50	super.setUncaughtExceptionHandler(handler);
51	if (useSystemClassLoader)
52	super.setContextClassLoader(ClassLoader.getSystemClassLoader());
53	}
54
55	/**
56	* Creates a ForkJoinWorkerThread operating in the given thread group and
57	* pool.
58	*
59	* @param group if non-null, the thread group for this thread
60	* @param pool the pool this thread works in
61	* @throws NullPointerException if pool is null
62	*/
63	/* TODO: protected */ ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool) {
64	this(group, pool, false, false);
65	}
66
67	/**
68	* Creates a ForkJoinWorkerThread operating in the given pool.
69	*
70	* @param pool the pool this thread works in
71	* @throws NullPointerException if pool is null
72	*/
73	protected ForkJoinWorkerThread(ForkJoinPool pool) {
74	this(null, pool, false, false);
75	}
76
77	/**
78	* Returns the pool hosting this thread.
79	*
80	* @return the pool
81	*/
82	public ForkJoinPool getPool() {
83	return pool;
84	}
85
86	/**
87	* Returns the unique index number of this thread in its pool.
88	* The returned value ranges from zero to the maximum number of
89	* threads (minus one) that may exist in the pool, and does not
90	* change during the lifetime of the thread. This method may be
91	* useful for applications that track status or collect results
92	* per-worker-thread rather than per-task.
93	*
94	* @return the index number
95	*/
96	public int getPoolIndex() {
97	return workQueue.getPoolIndex();
98	}
99
100	/**
101	* Initializes internal state after construction but before
102	* processing any tasks. If you override this method, you must
103	* invoke {@code super.onStart()} at the beginning of the method.
104	* Initialization requires care: Most fields must have legal
105	* default values, to ensure that attempted accesses from other
106	* threads work correctly even before this thread starts
107	* processing tasks.
108	*/
109	protected void onStart() {
110	}
111
112	/**
113	* Performs cleanup associated with termination of this worker
114	* thread. If you override this method, you must invoke
115	* {@code super.onTermination} at the end of the overridden method.
116	*
117	* @param exception the exception causing this thread to abort due
118	* to an unrecoverable error, or {@code null} if completed normally
119	*/
120	protected void onTermination(Throwable exception) {
121	}
122
123	/**
124	* This method is required to be public, but should never be
125	* called explicitly. It performs the main run loop to execute
126	* {@link ForkJoinTask}s.
127	*/
128	public void run() {
129	Throwable exception = null;
130	ForkJoinPool p = pool;
131	ForkJoinPool.WorkQueue w = workQueue;
132	if (p != null && w != null) { // skip on failed initialization
133	try {
134	p.registerWorker(w);
135	onStart();
136	p.runWorker(w);
137	} catch (Throwable ex) {
138	exception = ex;
139	} finally {
140	try {
141	onTermination(exception);
142	} catch (Throwable ex) {
143	if (exception == null)
144	exception = ex;
145	} finally {
146	p.deregisterWorker(this, exception);
147	}
148	}
149	}
150	}
151
152	/**
153	* A worker thread that has no permissions, is not a member of any
154	* user-defined ThreadGroup, uses the system class loader as
155	* thread context class loader, and erases all ThreadLocals after
156	* running each top-level task.
157	*/
158	static final class InnocuousForkJoinWorkerThread extends ForkJoinWorkerThread {
159	/** The ThreadGroup for all InnocuousForkJoinWorkerThreads */
160	private static final ThreadGroup innocuousThreadGroup =
161	AccessController.doPrivileged(new PrivilegedAction<>() {
162	public ThreadGroup run() {
163	ThreadGroup group = Thread.currentThread().getThreadGroup();
164	for (ThreadGroup p; (p = group.getParent()) != null; )
165	group = p;
166	return new ThreadGroup(
167	group, "InnocuousForkJoinWorkerThreadGroup");
168	}});
169
170	InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
171	super(innocuousThreadGroup, pool, true, true);
172	}
173
174	@Override // to silently fail
175	public void setUncaughtExceptionHandler(UncaughtExceptionHandler x) { }
176
177	@Override // paranoically
178	public void setContextClassLoader(ClassLoader cl) {
179	if (cl != null && ClassLoader.getSystemClassLoader() != cl)
180	throw new SecurityException("setContextClassLoader");
181	}
182	}
183	}