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