Attachment #26150 for bug #49159

View | Details | Raw Unified | Return to bug 49159
Collapse All | Expand All

Lines 28-33 Link Here

(-)conf/server.xml (+1 lines)
28	<!-- Prevent memory leaks due to use of particular java/javax APIs-->	28	<!-- Prevent memory leaks due to use of particular java/javax APIs-->
29	<Listener className="org.apache.catalina.core.JreMemoryLeakPreventionListener" />	29	<Listener className="org.apache.catalina.core.JreMemoryLeakPreventionListener" />
30	<Listener className="org.apache.catalina.mbeans.GlobalResourcesLifecycleListener" />	30	<Listener className="org.apache.catalina.mbeans.GlobalResourcesLifecycleListener" />
		31	<Listener className="org.apache.catalina.core.ThreadLocalLeakPreventionListener" />
31		32
32	<!-- Global JNDI resources	33	<!-- Global JNDI resources
33	Documentation at /docs/jndi-resources-howto.html	34	Documentation at /docs/jndi-resources-howto.html




     * default value of <code>false</code> will be used.
     */
    private boolean clearReferencesStopTimerThreads = false;

    /**
     * Should Tomcat renew the threads of the thread pool when the application
     * is stopped to avoid memory leaks because of uncleaned ThreadLocal
     * variables. This also requires that the threadRenewalDelay property of the
     * StandardThreadExecutor of ThreadPoolExecutor be set to a positive value.
     * objects is not performed in a thread-safe manner.
     */
    private boolean renewThreadsWhenStoppingContext = true;
    
    /**
     * Should the effective web.xml be logged when the context starts?

    }


    public boolean getRenewThreadsWhenStoppingContext() {
        return this.renewThreadsWhenStoppingContext;
    }
    public boolean getClearReferencesThreadLocals() {

    public void setRenewThreadsWhenStoppingContext(boolean renewThreadsWhenStoppingContext) {
        boolean oldRenewThreadsWhenStoppingContext =
            this.renewThreadsWhenStoppingContext;
        this.renewThreadsWhenStoppingContext = renewThreadsWhenStoppingContext;
        support.firePropertyChange("renewThreadsWhenStoppingContext",
                oldRenewThreadsWhenStoppingContext,
                this.renewThreadsWhenStoppingContext);
    }


    /**
     * Set the clearReferencesThreadLocals feature for this Context.
     *
     * @param clearReferencesThreadLocals The new flag value
     */
    public void setClearReferencesThreadLocals(
            boolean clearReferencesThreadLocals) {

        boolean oldClearReferencesThreadLocals =
            this.clearReferencesThreadLocals;
        this.clearReferencesThreadLocals = clearReferencesThreadLocals;
        support.firePropertyChange("clearReferencesStopThreads",
                                   oldClearReferencesThreadLocals,
                                   this.clearReferencesThreadLocals);

    }


Lines 84-89 Link Here

(-)java/org/apache/catalina/core/StandardThreadExecutor.java (+25 lines)
84	*/	84	*/
85	protected int maxQueueSize = Integer.MAX_VALUE;	85	protected int maxQueueSize = Integer.MAX_VALUE;
86		86
		87	/**
		88	* After a context is stopped, threads in the pool are renewed. To avoid
		89	* renewing all threads at the same time, this delay is observed between 2
		90	* threads being renewed.
		91	*/
		92	protected long threadRenewalDelay = 1000L;
		93
87	private TaskQueue taskqueue = null;	94	private TaskQueue taskqueue = null;
88	// ---------------------------------------------- Constructors	95	// ---------------------------------------------- Constructors
89	public StandardThreadExecutor() {	96	public StandardThreadExecutor() {
Lines 165-170 Link Here
165	}	172	}
166	} else throw new IllegalStateException("StandardThreadPool not started.");	173	} else throw new IllegalStateException("StandardThreadPool not started.");
167	}	174	}
		175
		176	public void contextStopping() {
		177	if (executor != null) {
		178	executor.contextStopping();
		179	}
		180	}
168		181
169	public int getThreadPriority() {	182	public int getThreadPriority() {
170	return threadPriority;	183	return threadPriority;
Lines 250-255 Link Here
250	return maxQueueSize;	263	return maxQueueSize;
251	}	264	}
252		265
		266	public long getThreadRenewalDelay() {
		267	return threadRenewalDelay;
		268	}
		269
		270	public void setThreadRenewalDelay(long threadRenewalDelay) {
		271	this.threadRenewalDelay = threadRenewalDelay;
		272	if(executor!=null) {
		273	executor.setThreadRenewalDelay(threadRenewalDelay);
		274	}
		275	}
		276
		277
253	// Statistics from the thread pool	278	// Statistics from the thread pool
254	@Override	279	@Override
255	public int getActiveCount() {	280	public int getActiveCount() {




/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.catalina.core;

import java.util.concurrent.Executor;

import org.apache.catalina.Container;
import org.apache.catalina.ContainerEvent;
import org.apache.catalina.ContainerListener;
import org.apache.catalina.Context;
import org.apache.catalina.Engine;
import org.apache.catalina.Host;
import org.apache.catalina.Lifecycle;
import org.apache.catalina.LifecycleEvent;
import org.apache.catalina.LifecycleListener;
import org.apache.catalina.Server;
import org.apache.catalina.Service;
import org.apache.catalina.connector.Connector;
import org.apache.coyote.ProtocolHandler;
import org.apache.coyote.ajp.AjpAprProtocol;
import org.apache.coyote.ajp.AjpProtocol;
import org.apache.coyote.http11.AbstractHttp11Protocol;
import org.apache.coyote.http11.Http11AprProtocol;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;

/**
 * A {@link LifecycleListener} that triggers the renewal of threads in Executor
 * pools when a {@link Context} is being stopped to avoid thread-local related
 * memory leaks.<br/>
 * Note : active threads will be renewed one by one when they come back to the
 * pool after executing their task, see
 * {@link org.apache.tomcat.util.threads.ThreadPoolExecutor}.afterExecute().<br/>
 * 
 * This listener must be declared in server.xml to be active.
 * 
 * @author slaurent
 * 
 */
public class ThreadLocalLeakPreventionListener implements LifecycleListener, ContainerListener {
    private static final Log log = LogFactory.getLog(ThreadLocalLeakPreventionListener.class);

    /**
     * Listens for {@link LifecycleEvent} for the start of the {@link Server} to
     * initialize itself and then for after_stop events of each {@link Context}.
     */
    @Override
    public void lifecycleEvent(LifecycleEvent event) {
        try {
            Lifecycle lifecycle = event.getLifecycle();
            if (Lifecycle.AFTER_START_EVENT.equals(event.getType()) && lifecycle instanceof Server) {
                // when the server starts, we register ourself as listener for
                // all context
                // as well as container event listener so that we know when new
                // Context are deployed
                Server server = (Server) lifecycle;
                registerListenersForServer(server);
            }

            if (Lifecycle.AFTER_STOP_EVENT.equals(event.getType()) && lifecycle instanceof Context) {
                stopIdleThreads((Context) lifecycle);
            }
        } catch (Exception e) {
            log.error("Exception processing event " + event, e);
        }
    }

    @Override
    public void containerEvent(ContainerEvent event) {
        try {
            String type = event.getType();
            if (Container.ADD_CHILD_EVENT.equals(type)) {
                processContainerAddChild(event.getContainer(), (Container) event.getData());
            } else if (Container.REMOVE_CHILD_EVENT.equals(type)) {
                processContainerRemoveChild(event.getContainer(), (Container) event.getData());
            }
        } catch (Exception e) {
            log.error("Exception processing event " + event, e);
        }

    }

    private void registerListenersForServer(Server server) {
        for (Service service : server.findServices()) {
            Engine engine = (Engine) service.getContainer();
            engine.addContainerListener(this);
            registerListenersForEngine(engine);
        }

    }

    private void registerListenersForEngine(Engine engine) {
        for (Container hostContainer : engine.findChildren()) {
            Host host = (Host) hostContainer;
            host.addContainerListener(this);
            registerListenersForHost(host);
        }
    }

    private void registerListenersForHost(Host host) {
        for (Container contextContainer : host.findChildren()) {
            Context context = (Context) contextContainer;
            registerContextListener(context);
        }
    }

    private void registerContextListener(Context context) {
        context.addLifecycleListener(this);
    }

    protected void processContainerAddChild(Container parent, Container child) {
        if (log.isDebugEnabled())
            log.debug("Process addChild[parent=" + parent + ",child=" + child + "]");

        try {
            if (child instanceof Context) {
                registerContextListener((Context) child);
            } else if (child instanceof Engine) {
                registerListenersForEngine((Engine) child);
            } else if (child instanceof Host) {
                registerListenersForHost((Host) child);
            }
        } catch (Throwable t) {
            log.error("processContainerAddChild: Throwable", t);
        }

    }

    protected void processContainerRemoveChild(Container parent, Container child) {

        if (log.isDebugEnabled())
            log.debug("Process removeChild[parent=" + parent + ",child=" + child + "]");

        try {
            if (child instanceof Context) {
                Context context = (Context) child;
                context.removeLifecycleListener(this);
            } else if (child instanceof Host) {
                Host host = (Host) child;
                host.removeContainerListener(this);
            } else if (child instanceof Engine) {
                Engine engine = (Engine) child;
                engine.removeContainerListener(this);
            }
        } catch (Throwable t) {
            log.error("processContainerRemoveChild: Throwable", t);
        }

    }

    /**
     * Updates each ThreadPoolExecutor with the current time, which is the time
     * when a context is being stopped.
     * 
     * @param context
     *            the context being stopped, used to discover all the Connectors
     *            of its parent Service.
     */
    private void stopIdleThreads(Context context) {
        if (context instanceof StandardContext && !((StandardContext) context).getRenewThreadsWhenStoppingContext()) {
            log.debug("Not renewing threads when the context is stopping, it is configured not to do it.");
            return;
        }

        Engine engine = (Engine) context.getParent().getParent();
        Service service = engine.getService();
        Connector[] connectors = service.findConnectors();
        if (connectors != null) {
            for (Connector connector : connectors) {
                ProtocolHandler handler = connector.getProtocolHandler();
                Executor executor = null;
                if (handler instanceof AbstractHttp11Protocol) {
                    executor = ((AbstractHttp11Protocol) handler).getExecutor();
                } else if (handler instanceof AjpProtocol) {
                    executor = ((AjpProtocol) handler).getExecutor();
                } else if (handler instanceof AjpAprProtocol) {
                    executor = ((AjpAprProtocol) handler).getExecutor();
                } else if (handler instanceof Http11AprProtocol) {
                    executor = ((Http11AprProtocol) handler).getExecutor();
                }

                if (executor instanceof ThreadPoolExecutor) {
                    ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
                    threadPoolExecutor.contextStopping();
                } else if (executor instanceof StandardThreadExecutor) {
                    StandardThreadExecutor stdThreadExecutor = (StandardThreadExecutor) executor;
                    stdThreadExecutor.contextStopping();
                }

            }
        }
    }
}

Lines 115-124 Link Here

(-)java/org/apache/catalina/core/mbeans-descriptors.xml (-4 / +9 lines)
115	description="Should Tomcat attempt to terminate threads that have been started by the web application? Advisable to be used only in a development environment."	115	description="Should Tomcat attempt to terminate threads that have been started by the web application? Advisable to be used only in a development environment."
116	type="boolean"/>	116	type="boolean"/>
117		117
118	<attribute name="clearReferencesThreadLocals"
119	description="Should Tomcat attempt to clear any ThreadLocal objects that are instances of classes loaded by this class loader. "
120	type="boolean"/>
121
122	<attribute name="clearReferencesStopTimerThreads"	118	<attribute name="clearReferencesStopTimerThreads"
123	description="Should Tomcat attempt to terminate TimerThreads that have been started by the web application? Advisable to be used only in a development environment."	119	description="Should Tomcat attempt to terminate TimerThreads that have been started by the web application? Advisable to be used only in a development environment."
124	type="boolean"/>	120	type="boolean"/>
Lines 272-277 Link Here
272	description="The reloadable flag for this web application"	268	description="The reloadable flag for this web application"
273	type="boolean"/>	269	type="boolean"/>
274		270
		271	<attribute name="renewThreadsWhenStoppingContext"
		272	description="Should Tomcat renew the threads of the thread pool when the application is stopped to avoid memory leaks because of uncleaned ThreadLocal variables."
		273	type="boolean"/>
		274
275	<attribute name="saveConfig"	275	<attribute name="saveConfig"
276	description="Should the configuration be written as needed on startup"	276	description="Should the configuration be written as needed on startup"
277	is="true"	277	is="true"
Lines 1521-1526 Link Here
1521	<attribute name="threadPriority"	1521	<attribute name="threadPriority"
1522	description="The thread priority for threads in this thread pool"	1522	description="The thread priority for threads in this thread pool"
1523	type="int"/>	1523	type="int"/>
		1524
		1525	<attribute name="threadRenewalDelay"
		1526	description="After a context is stopped, threads in the pool are renewed. To avoid renewing all threads at the same time, this delay is observed between 2 threads being renewed. Value is in ms, default value is 1000ms. If negative, threads are not renewed."
		1527	type="long"/>
		1528
1524	</mbean>	1529	</mbean>
1525		1530
1526	<mbean name="StandardWrapper"	1531	<mbean name="StandardWrapper"

Lines 44-54 Link Here

(-)java/org/apache/catalina/loader/LocalStrings.properties (-5 / +3 lines)
44	webappClassLoader.clearReferencesResourceBundlesFail=Failed to clear ResourceBundle references for web application [{0}]	44	webappClassLoader.clearReferencesResourceBundlesFail=Failed to clear ResourceBundle references for web application [{0}]
45	webappClassLoader.clearRmiInfo=Failed to find class sun.rmi.transport.Target to clear context class loader for web application [{0}]. This is expected on non-Sun JVMs.	45	webappClassLoader.clearRmiInfo=Failed to find class sun.rmi.transport.Target to clear context class loader for web application [{0}]. This is expected on non-Sun JVMs.
46	webappClassLoader.clearRmiFail=Failed to clear context class loader referenced from sun.rmi.transport.Target for web application [{0}]	46	webappClassLoader.clearRmiFail=Failed to clear context class loader referenced from sun.rmi.transport.Target for web application [{0}]
47	webappClassLoader.clearThreadLocalDebug=The web application [{0}] created a ThreadLocal with key of type [{1}] (value [{2}]). The ThreadLocal has been correctly set to null and the key will be removed by GC.	47	webappClassLoader.checkThreadLocalsForLeaksDebug=The web application [{0}] created a ThreadLocal with key of type [{1}] (value [{2}]). The ThreadLocal has been correctly set to null and the key will be removed by GC.
48	webappClassLoader.clearThreadLocal=The web application [{0}] created a ThreadLocal with key of type [{1}] (value [{2}]) and a value of type [{3}] (value [{4}]) but failed to remove it when the web application was stopped. This is very likely to create a memory leak.	48	webappClassLoader.checkThreadLocalsForLeaks=The web application [{0}] created a ThreadLocal with key of type [{1}] (value [{2}]) and a value of type [{3}] (value [{4}]) but failed to remove it when the web application was stopped. This is very likely to create a memory leak, but threads are going to be renewed over time so that the leak should be mitigated.
49	webappClassLoader.clearThreadLocalDebugClear=To simplify the process of tracing memory leaks, the key has been forcibly removed.	49	webappClassLoader.checkThreadLocalsForLeaksFail=Failed to check for ThreadLocal references for web application [{0}]
50	webappClassLoader.clearThreadLocalClear=To prevent a memory leak, the ThreadLocal has been forcibly removed.
51	webappClassLoader.clearThreadLocalFail=Failed to clear ThreadLocal references for web application [{0}]
52	webappClassLoader.stopThreadFail=Failed to terminate thread named [{0}] for web application [{1}]	50	webappClassLoader.stopThreadFail=Failed to terminate thread named [{0}] for web application [{1}]
53	webappClassLoader.stopTimerThreadFail=Failed to terminate TimerThread named [{0}] for web application [{1}]	51	webappClassLoader.stopTimerThreadFail=Failed to terminate TimerThread named [{0}] for web application [{1}]
54	webappClassLoader.validationErrorJarPath=Unable to validate JAR entry with name {0}	52	webappClassLoader.validationErrorJarPath=Unable to validate JAR entry with name {0}




    private boolean clearReferencesStopTimerThreads = false;

    /**
     * Should Tomcat attempt to clear any ThreadLocal objects that are instances
     * of classes loaded by this class loader. Failure to remove any such
     * objects will result in a memory leak on web application stop, undeploy or
     * reload. It is disabled by default since the clearing of the ThreadLocal
     * objects is not performed in a thread-safe manner.
     */
    private boolean clearReferencesThreadLocals = false;
    
    /**
     * Should Tomcat call {@link org.apache.juli.logging.LogFactory#release()}
     * when the class loader is stopped? If not specified, the default value
     * of <code>true</code> is used. Changing the default setting is likely to

     }


     /**
      * Return the clearReferencesThreadLocals flag for this Context.
      */
     public boolean getClearReferencesThreadLocals() {
         return (this.clearReferencesThreadLocals);
     }


     /**
      * Set the clearReferencesThreadLocals feature for this Context.
      *
      * @param clearReferencesThreadLocals The new flag value
      */
     public void setClearReferencesThreadLocals(
             boolean clearReferencesThreadLocals) {
         this.clearReferencesThreadLocals = clearReferencesThreadLocals;
     }


    // ------------------------------------------------------- Reloader Methods



        clearReferencesThreads();
        
        // Clear any ThreadLocals loaded by this class loader
        checkThreadLocalsForLeaks();
        
        // Clear RMI Targets loaded by this class loader
        clearReferencesRmiTargets();

        }
    }

    private void checkThreadLocalsForLeaks() {
        Thread[] threads = getThreads();

        try {

                if (threads[i] != null) {
                    // Clear the first map
                    threadLocalMap = threadLocalsField.get(threads[i]);
                    checkThreadLocalMapForLeaks(threadLocalMap, tableField);
                    // Clear the second map
                    threadLocalMap =
                        inheritableThreadLocalsField.get(threads[i]);
                    checkThreadLocalMapForLeaks(threadLocalMap, tableField);
                }
            }
        } catch (SecurityException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        } catch (NoSuchFieldException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        } catch (ClassNotFoundException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        } catch (IllegalArgumentException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        } catch (IllegalAccessException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        } catch (NoSuchMethodException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        } catch (InvocationTargetException e) {
            log.warn(sm.getString("webappClassLoader.checkThreadLocalsForLeaksFail",
                    contextName), e);
        }       
    }


    /*
     * Analyzes the given thread local map object. Also pass in the field that
     * points to the internal table to save re-calculating it on every
     * call to this method.
     */
    private void checkThreadLocalMapForLeaks(Object map, Field internalTableField)
            throws NoSuchMethodException, IllegalAccessException,
            NoSuchFieldException, InvocationTargetException {
        if (map != null) {
            Method mapRemove =
                map.getClass().getDeclaredMethod("remove",
                        ThreadLocal.class);
            mapRemove.setAccessible(true);
            Object[] table = (Object[]) internalTableField.get(map);
            int staleEntriesCount = 0;
            if (table != null) {
                for (int j =0; j < table.length; j++) {
                    if (table[j] != null) {
                        boolean potentialLeak = false;
                        // Check the key
                        Object key = ((Reference<?>) table[j]).get();
                        if (this.equals(key) || objectIsLoadedByThisOrChildClassLoader(key)) {
                            potentialLeak = true;
                            remove = true;
                        }
                        // Check the value
                        Field valueField =
                            table[j].getClass().getDeclaredField("value");
                        valueField.setAccessible(true);
                        Object value = valueField.get(table[j]);
                        if (this.equals(value) || objectIsLoadedByThisOrChildClassLoader(value)) {
                            potentialLeak = true;
                            remove = true;
                        }
                        if (potentialLeak) {
                            Object[] args = new Object[5];
                            args[0] = contextName;
                            if (key != null) {

                            if (value != null) {
                                args[3] = value.getClass().getCanonicalName();
                                args[4] = value.toString();
                                log.error(sm.getString(
                                        "webappClassLoader.checkThreadLocalsForLeaks",
                                        args));
                            } else {
                                if (log.isDebugEnabled()) {
                                    log.debug(sm.getString(
                                            "webappClassLoader.checkThreadLocalsForLeaksDebug",
                                            args));
                                    if (clearReferencesThreadLocals) {
                                        log.debug(sm.getString(
                                                "webappClassLoader.clearThreadLocalDebugClear"));
                                    }
                                }
                            } else {
                                log.error(sm.getString(
                                        "webappClassLoader.clearThreadLocal",
                                        args));
                                if (clearReferencesThreadLocals) {
                                    log.info(sm.getString(
                                            "webappClassLoader.clearThreadLocalClear"));
                                }
                            }
                            if (clearReferencesThreadLocals) {
                                if (key == null) {
                                  staleEntriesCount++;
                                } else {
                                  mapRemove.invoke(map, key);
                                }
                            }
                        }
                    }
                }
            }
            if (staleEntriesCount > 0) {
                Method mapRemoveStale =
                    map.getClass().getDeclaredMethod("expungeStaleEntries");
                mapRemoveStale.setAccessible(true);
                mapRemoveStale.invoke(map);
            }
        }
    }


    }


    private boolean objectIsLoadedByThisOrChildClassLoader(Object o) {
    	if(o == null) {
    		return false;
    	}

    	Class<?> clazz = o.getClass();
    	if(o instanceof Class) {
    		clazz = (Class<?>)clazz;
    	}
    	
    	for(ClassLoader cl = clazz.getClassLoader(); cl != null; cl = cl.getParent()) {
    		if(cl == this) {
    			return true;
    		}
    	}
    	return false;
    }
    /**
     * Determine whether a class was loaded by this class loader or one of
     * its child class loaders.

Lines 592-599 Link Here

(-)java/org/apache/catalina/loader/WebappLoader.java (-2 lines)
592	((StandardContext) container).getClearReferencesStopThreads());	592	((StandardContext) container).getClearReferencesStopThreads());
593	classLoader.setClearReferencesStopTimerThreads(	593	classLoader.setClearReferencesStopTimerThreads(
594	((StandardContext) container).getClearReferencesStopTimerThreads());	594	((StandardContext) container).getClearReferencesStopTimerThreads());
595	classLoader.setClearReferencesThreadLocals(
596	((StandardContext) container).getClearReferencesThreadLocals());
597	}	595	}
598		596
599	for (int i = 0; i < repositories.length; i++) {	597	for (int i = 0; i < repositories.length; i++) {

Lines 34-39 Link Here

(-)java/org/apache/tomcat/util/threads/TaskQueue.java (+44 lines)
34	private static final long serialVersionUID = 1L;	34	private static final long serialVersionUID = 1L;
35		35
36	private ThreadPoolExecutor parent = null;	36	private ThreadPoolExecutor parent = null;
		37
		38	// no need to be volatile, the one times when we change and read it occur in
		39	// a single thread (the one that did stop a context and fired listeners)
		40	private Integer forcedRemainingCapacity = null;
37		41
38	public TaskQueue() {	42	public TaskQueue() {
39	super();	43	super();
Lines 74-77 Link Here
74	//if we reached here, we need to add it to the queue	78	//if we reached here, we need to add it to the queue
75	return super.offer(o);	79	return super.offer(o);
76	}	80	}
		81
		82
		83	@Override
		84	public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
		85	Runnable runnable = super.poll(timeout, unit);
		86	if (runnable == null && parent != null) {
		87	// the poll timed out, it gives an opportunity to stop the current
		88	// thread if needed to avoid memory leaks.
		89	parent.stopCurrentThreadIfNeeded();
		90	}
		91	return runnable;
		92	}
		93
		94
		95	@Override
		96	public Runnable take() throws InterruptedException {
		97	if (parent != null && parent.currentThreadShouldBeStopped()) {
		98	return poll(parent.getKeepAliveTime(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
		99	//yes, this may return null (in case of timeout) which normally does not occur with take()
		100	//but the ThreadPoolExecutor implementation allows this
		101	}
		102	return super.take();
		103	}
		104
		105	@Override
		106	public int remainingCapacity() {
		107	if(forcedRemainingCapacity != null) {
		108	// ThreadPoolExecutor.setCorePoolSize checks that
		109	// remainingCapacity==0 to allow to interrupt idle threads
		110	// I don't see why, but this hack allows to conform to this
		111	// "requirement"
		112	return forcedRemainingCapacity.intValue();
		113	}
		114	return super.remainingCapacity();
		115	}
		116
		117	public void setForcedRemainingCapacity(Integer forcedRemainingCapacity) {
		118	this.forcedRemainingCapacity = forcedRemainingCapacity;
		119	}
		120
77	}	121	}




/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.tomcat.util.threads;

/**
 * A Thread implementation that records the time at which it was created.
 * 
 * @author slaurent
 * 
 */
public class TaskThread extends Thread {

	private final long creationTime;

	public TaskThread(ThreadGroup group, Runnable target, String name) {
		super(group, target, name);
		this.creationTime = System.currentTimeMillis();
	}

	public TaskThread(ThreadGroup group, Runnable target, String name,
			long stackSize) {
		super(group, target, name, stackSize);
		this.creationTime = System.currentTimeMillis();
	}

	/**
	 * @return the time (in ms) at which this thread was created
	 */
	public final long getCreationTime() {
		return creationTime;
	}

}

Lines 39-45 Link Here

(-)java/org/apache/tomcat/util/threads/TaskThreadFactory.java (-1 / +1 lines)
39		39
40	@Override	40	@Override
41	public Thread newThread(Runnable r) {	41	public Thread newThread(Runnable r) {
42	Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement());	42	TaskThread t = new TaskThread(group, r, namePrefix + threadNumber.getAndIncrement());
43	t.setDaemon(daemon);	43	t.setDaemon(daemon);
44	t.setPriority(threadPriority);	44	t.setPriority(threadPriority);
45	return t;	45	return t;




 */
package org.apache.tomcat.util.threads;

import java.lang.Thread.UncaughtExceptionHandler;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;

/**
 * Same as a java.util.concurrent.ThreadPoolExecutor but implements a much more efficient
 * {@link #getSubmittedCount()} method, to be used to properly handle the work queue.

 *
 */
public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
    private static final Log log = LogFactory.getLog(ThreadPoolExecutor.class);

    /**
     * The number of tasks submitted but not yet finished. This includes tasks
     * in the queue and tasks that have been handed to a worker thread but the

     * This number is always greater or equal to {@link #getActiveCount()}.
     */
    private final AtomicInteger submittedCount = new AtomicInteger(0);
    private final AtomicLong lastContextStoppedTime = new AtomicLong(0L);

    /**
     * Most recent time in ms when a thread decided to kill itself to avoid
     * potential memory leaks. Useful to throttle the rate of renewals of
     * threads.
     */
    private final AtomicLong lastTimeThreadKilledItself = new AtomicLong(0L);

    /**
     * Delay in ms between 2 threads being renewed. If negative, do not renew threads.
     */
    private long threadRenewalDelay = 1000L;

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new RejectHandler());
    }
    
    public long getThreadRenewalDelay() {
        return threadRenewalDelay;
    }

    public void setThreadRenewalDelay(long threadRenewalDelay) {
        this.threadRenewalDelay = threadRenewalDelay;
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        submittedCount.decrementAndGet();

        if (t == null) {
            stopCurrentThreadIfNeeded();
        }
    }

    /**
     * If the current thread was started before the last time when a context was
     * stopped, an exception is thrown so that the current thread is stopped.
     */
    protected void stopCurrentThreadIfNeeded() {
        if (currentThreadShouldBeStopped()) {
            long lastTime = lastTimeThreadKilledItself.longValue();
            if (lastTime + threadRenewalDelay < System.currentTimeMillis()) {
                if (lastTimeThreadKilledItself.compareAndSet(lastTime, System.currentTimeMillis() + 1)) {
                    // OK, it's really time to dispose of this thread
                    
                    final String msg = "Stopping thread " + Thread.currentThread().getName()
                            + " to avoid potential memory leaks after a context was stopped.";
                    Thread.currentThread().setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
                        @Override
                        public void uncaughtException(Thread t, Throwable e) {
                            // yes, swallow the exception
                            log.debug(msg);
                        }
                    });
                    throw new RuntimeException(msg);
                }
            }
        }
    }
    
    protected boolean currentThreadShouldBeStopped() {
        if (threadRenewalDelay >= 0 && Thread.currentThread() instanceof TaskThread) {
            TaskThread currentTaskThread = (TaskThread) Thread.currentThread();
            if (currentTaskThread.getCreationTime() < this.lastContextStoppedTime.longValue()) {
                return true;
            }
        }
        return false;
    }

    public int getSubmittedCount() {
        return submittedCount.get();
    }

            
        }
    }

    public void contextStopping() {
        this.lastContextStoppedTime.set(System.currentTimeMillis());

        // save the current pool parameters to restore them later
        int savedCorePoolSize = this.getCorePoolSize();
        TaskQueue taskQueue = getQueue() instanceof TaskQueue ? (TaskQueue) getQueue() : null;
        if (taskQueue != null) {
            // note by slaurent : quite oddly threadPoolExecutor.setCorePoolSize
            // checks that queue.remainingCapacity()==0. I did not understand
            // why, but to get the intended effect of waking up idle threads, I
            // temporarily fake this condition.
            taskQueue.setForcedRemainingCapacity(0);
        }

        // setCorePoolSize(0) wakes idle threads
        this.setCorePoolSize(0);

        // wait a little so that idle threads wake and poll the queue again,
        // this time always with a timeout (queue.poll() instead of queue.take())
        // even if we did not wait enough, TaskQueue.take() takes care of timing out
        // so that we are sure that all threads of the pool are renewed in a limited
        // time, something like (threadKeepAlive + longest request time)
        try {
            Thread.sleep(200L);
        } catch (InterruptedException e) {
            //yes, ignore
        }
        
        if (taskQueue != null) {
            // ok, restore the state of the queue and pool
            taskQueue.setForcedRemainingCapacity(null);
        }
        this.setCorePoolSize(savedCorePoolSize);
    }
    
    private static class RejectHandler implements RejectedExecutionHandler {
        @Override




        not specified, the default value of <code>false</code> will be used.</p>
      </attribute>

      <attribute name="renewThreadsWhenStoppingContext" required="false">
        <p>If <code>true</code>, when this context is stopped, Tomcat renews all
        the threads from the thread pool that was used to serve this context.
        This also requires that the 
        <code>ThreadLocalLeakPreventionListener</code> be configured in 
        <code>server.xml</code> and that the <code>threadRenewalDelay</code>
        property of the <code>Executor</code> be &gt;=0. If not specified, the 
        default value of <code>true</code> will be used.</p>
      </attribute>

      <attribute name="processTlds" required="false">

Lines 106-111 Link Here

(-)webapps/docs/config/executor.xml (+5 lines)
106	<p>(boolean) Whether minSpareThreads should be started when starting the Executor or not,	106	<p>(boolean) Whether minSpareThreads should be started when starting the Executor or not,
107	the default is <code>false</code></p>	107	the default is <code>false</code></p>
108	</attribute>	108	</attribute>
		109	<attribute name="threadRenewalDelay" required="false">
		110	<p>After a context is stopped, threads in the pool are renewed. To avoid renewing all threads at the same time,
		111	this delay is observed between 2 threads being renewed. Value is in ms, default value is 1000ms.
		112	If negative, threads are not renewed.</p>
		113	</attribute>
109	</attributes>	114	</attributes>
110		115
111		116

Return to bug 49159

Lines 775-789 Link Here

(-)java/org/apache/catalina/core/StandardContext.java (-25 / +16 lines)
775	* default value of <code>false</code> will be used.	775	* default value of <code>false</code> will be used.
776	*/	776	*/
777	private boolean clearReferencesStopTimerThreads = false;	777	private boolean clearReferencesStopTimerThreads = false;
778		778
779	/**	779	/**
780	* Should Tomcat attempt to clear any ThreadLocal objects that are instances	780	* Should Tomcat renew the threads of the thread pool when the application
781	* of classes loaded by this class loader. Failure to remove any such	781	* is stopped to avoid memory leaks because of uncleaned ThreadLocal
782	* objects will result in a memory leak on web application stop, undeploy or	782	* variables. This also requires that the threadRenewalDelay property of the
783	* reload. It is disabled by default since the clearing of the ThreadLocal	783	* StandardThreadExecutor of ThreadPoolExecutor be set to a positive value.
784	* objects is not performed in a thread-safe manner.
785	*/	784	*/
786	private boolean clearReferencesThreadLocals = false;	785	private boolean renewThreadsWhenStoppingContext = true;
787		786
788	/**	787	/**
789	* Should the effective web.xml be logged when the context starts?	788	* Should the effective web.xml be logged when the context starts?
Lines 2351-2380 Link Here
2351	}	2350	}
2352		2351
2353		2352
2354	/**	2353	public boolean getRenewThreadsWhenStoppingContext() {
2355	* Return the clearReferencesThreadLocals flag for this Context.	2354	return this.renewThreadsWhenStoppingContext;
2356	*/	2355	}
2357	public boolean getClearReferencesThreadLocals() {
2358		2356
2359	return (this.clearReferencesThreadLocals);	2357	public void setRenewThreadsWhenStoppingContext(boolean renewThreadsWhenStoppingContext) {
2360		2358	boolean oldRenewThreadsWhenStoppingContext =
		2359	this.renewThreadsWhenStoppingContext;
		2360	this.renewThreadsWhenStoppingContext = renewThreadsWhenStoppingContext;
		2361	support.firePropertyChange("renewThreadsWhenStoppingContext",
		2362	oldRenewThreadsWhenStoppingContext,
		2363	this.renewThreadsWhenStoppingContext);
2361	}	2364	}
2362		2365
2363
2364	/**
2365	* Set the clearReferencesThreadLocals feature for this Context.
2366	*
2367	* @param clearReferencesThreadLocals The new flag value
2368	*/
2369	public void setClearReferencesThreadLocals(	2366	public void setClearReferencesThreadLocals(
2370	boolean clearReferencesThreadLocals) {	2367	boolean clearReferencesThreadLocals) {
2371		2368
2372	boolean oldClearReferencesThreadLocals =
2373	this.clearReferencesThreadLocals;
2374	this.clearReferencesThreadLocals = clearReferencesThreadLocals;
2375	support.firePropertyChange("clearReferencesStopThreads",
2376	oldClearReferencesThreadLocals,
2377	this.clearReferencesThreadLocals);
2378		2369
2379	}	2370	}
2380		2371

Lines 441-453 Link Here

(-)webapps/docs/config/context.xml (-7 / +8 lines)
441	not specified, the default value of <code>false</code> will be used.</p>	441	not specified, the default value of <code>false</code> will be used.</p>
442	</attribute>	442	</attribute>
443		443
444	<attribute name="clearReferencesThreadLocals" required="false">	444	<attribute name="renewThreadsWhenStoppingContext" required="false">
445	<p>If <code>true</code>, Tomcat attempts to clear any ThreadLocal	445	<p>If <code>true</code>, when this context is stopped, Tomcat renews all
446	objects that are instances of classes loaded by this class loader.	446	the threads from the thread pool that was used to serve this context.
447	Failure to remove any such objects will result in a memory leak on web	447	This also requires that the
448	application stop, undeploy or reload. If not specified, the default	448	<code>ThreadLocalLeakPreventionListener</code> be configured in
449	value of <code>false</code> will be used since the clearing of the	449	<code>server.xml</code> and that the <code>threadRenewalDelay</code>
450	ThreadLocal objects is not performed in a thread-safe manner.</p>	450	property of the <code>Executor</code> be >=0. If not specified, the
		451	default value of <code>true</code> will be used.</p>
451	</attribute>	452	</attribute>
452		453
453	<attribute name="processTlds" required="false">	454	<attribute name="processTlds" required="false">

Line 0 Link Here

(-)java/org/apache/catalina/core/ThreadLocalLeakPreventionListener.java (+209 lines)
		1	/*
		2	* Licensed to the Apache Software Foundation (ASF) under one or more
		3	* contributor license agreements. See the NOTICE file distributed with
		4	* this work for additional information regarding copyright ownership.
		5	* The ASF licenses this file to You under the Apache License, Version 2.0
		6	* (the "License"); you may not use this file except in compliance with
		7	* the License. You may obtain a copy of the License at
		8	*
		9	* http://www.apache.org/licenses/LICENSE-2.0
		10	*
		11	* Unless required by applicable law or agreed to in writing, software
		12	* distributed under the License is distributed on an "AS IS" BASIS,
		13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
		14	* See the License for the specific language governing permissions and
		15	* limitations under the License.
		16	*/
		17
		18	package org.apache.catalina.core;
		19
		20	import java.util.concurrent.Executor;
		21
		22	import org.apache.catalina.Container;
		23	import org.apache.catalina.ContainerEvent;
		24	import org.apache.catalina.ContainerListener;
		25	import org.apache.catalina.Context;
		26	import org.apache.catalina.Engine;
		27	import org.apache.catalina.Host;
		28	import org.apache.catalina.Lifecycle;
		29	import org.apache.catalina.LifecycleEvent;
		30	import org.apache.catalina.LifecycleListener;
		31	import org.apache.catalina.Server;
		32	import org.apache.catalina.Service;
		33	import org.apache.catalina.connector.Connector;
		34	import org.apache.coyote.ProtocolHandler;
		35	import org.apache.coyote.ajp.AjpAprProtocol;
		36	import org.apache.coyote.ajp.AjpProtocol;
		37	import org.apache.coyote.http11.AbstractHttp11Protocol;
		38	import org.apache.coyote.http11.Http11AprProtocol;
		39	import org.apache.juli.logging.Log;
		40	import org.apache.juli.logging.LogFactory;
		41	import org.apache.tomcat.util.threads.ThreadPoolExecutor;
		42
		43	/**
		44	* A {@link LifecycleListener} that triggers the renewal of threads in Executor
		45	* pools when a {@link Context} is being stopped to avoid thread-local related
		46	* memory leaks.<br/>
		47	* Note : active threads will be renewed one by one when they come back to the
		48	* pool after executing their task, see
		49	* {@link org.apache.tomcat.util.threads.ThreadPoolExecutor}.afterExecute().<br/>
		50	*
		51	* This listener must be declared in server.xml to be active.
		52	*
		53	* @author slaurent
		54	*
		55	*/
		56	public class ThreadLocalLeakPreventionListener implements LifecycleListener, ContainerListener {
		57	private static final Log log = LogFactory.getLog(ThreadLocalLeakPreventionListener.class);
		58
		59	/**
		60	* Listens for {@link LifecycleEvent} for the start of the {@link Server} to
		61	* initialize itself and then for after_stop events of each {@link Context}.
		62	*/
		63	@Override
		64	public void lifecycleEvent(LifecycleEvent event) {
65	try {
66	Lifecycle lifecycle = event.getLifecycle();
67	if (Lifecycle.AFTER_START_EVENT.equals(event.getType()) && lifecycle instanceof Server) {
68	// when the server starts, we register ourself as listener for
69	// all context
70	// as well as container event listener so that we know when new
71	// Context are deployed
72	Server server = (Server) lifecycle;
73	registerListenersForServer(server);
74	}
75
76	if (Lifecycle.AFTER_STOP_EVENT.equals(event.getType()) && lifecycle instanceof Context) {
77	stopIdleThreads((Context) lifecycle);
78	}
79	} catch (Exception e) {
80	log.error("Exception processing event " + event, e);
81	}
82	}
83
84	@Override
85	public void containerEvent(ContainerEvent event) {
86	try {
87	String type = event.getType();
88	if (Container.ADD_CHILD_EVENT.equals(type)) {
89	processContainerAddChild(event.getContainer(), (Container) event.getData());
90	} else if (Container.REMOVE_CHILD_EVENT.equals(type)) {
91	processContainerRemoveChild(event.getContainer(), (Container) event.getData());
92	}
93	} catch (Exception e) {
94	log.error("Exception processing event " + event, e);
95	}
96
97	}
98
99	private void registerListenersForServer(Server server) {
100	for (Service service : server.findServices()) {
101	Engine engine = (Engine) service.getContainer();
102	engine.addContainerListener(this);
103	registerListenersForEngine(engine);
104	}
105
106	}
107
108	private void registerListenersForEngine(Engine engine) {
109	for (Container hostContainer : engine.findChildren()) {
110	Host host = (Host) hostContainer;
111	host.addContainerListener(this);
112	registerListenersForHost(host);
113	}
114	}
115
116	private void registerListenersForHost(Host host) {
117	for (Container contextContainer : host.findChildren()) {
118	Context context = (Context) contextContainer;
119	registerContextListener(context);
120	}
121	}
122
123	private void registerContextListener(Context context) {
124	context.addLifecycleListener(this);
125	}
126
127	protected void processContainerAddChild(Container parent, Container child) {
128	if (log.isDebugEnabled())
129	log.debug("Process addChild[parent=" + parent + ",child=" + child + "]");
130
131	try {
132	if (child instanceof Context) {
133	registerContextListener((Context) child);
134	} else if (child instanceof Engine) {
135	registerListenersForEngine((Engine) child);
136	} else if (child instanceof Host) {
137	registerListenersForHost((Host) child);
138	}
139	} catch (Throwable t) {
140	log.error("processContainerAddChild: Throwable", t);
141	}
142
143	}
144
145	protected void processContainerRemoveChild(Container parent, Container child) {
146
147	if (log.isDebugEnabled())
148	log.debug("Process removeChild[parent=" + parent + ",child=" + child + "]");
149
150	try {
151	if (child instanceof Context) {
152	Context context = (Context) child;
153	context.removeLifecycleListener(this);
154	} else if (child instanceof Host) {
155	Host host = (Host) child;
156	host.removeContainerListener(this);
157	} else if (child instanceof Engine) {
158	Engine engine = (Engine) child;
159	engine.removeContainerListener(this);
160	}
161	} catch (Throwable t) {
162	log.error("processContainerRemoveChild: Throwable", t);
163	}
164
165	}
166
167	/**
168	* Updates each ThreadPoolExecutor with the current time, which is the time
169	* when a context is being stopped.
170	*
171	* @param context
172	* the context being stopped, used to discover all the Connectors
173	* of its parent Service.
174	*/
175	private void stopIdleThreads(Context context) {
176	if (context instanceof StandardContext && !((StandardContext) context).getRenewThreadsWhenStoppingContext()) {
177	log.debug("Not renewing threads when the context is stopping, it is configured not to do it.");
178	return;
179	}
180
181	Engine engine = (Engine) context.getParent().getParent();
182	Service service = engine.getService();
183	Connector[] connectors = service.findConnectors();
184	if (connectors != null) {
185	for (Connector connector : connectors) {
186	ProtocolHandler handler = connector.getProtocolHandler();
187	Executor executor = null;
188	if (handler instanceof AbstractHttp11Protocol) {
189	executor = ((AbstractHttp11Protocol) handler).getExecutor();
190	} else if (handler instanceof AjpProtocol) {
191	executor = ((AjpProtocol) handler).getExecutor();
192	} else if (handler instanceof AjpAprProtocol) {
193	executor = ((AjpAprProtocol) handler).getExecutor();
194	} else if (handler instanceof Http11AprProtocol) {
195	executor = ((Http11AprProtocol) handler).getExecutor();
196	}
197
198	if (executor instanceof ThreadPoolExecutor) {
199	ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
200	threadPoolExecutor.contextStopping();
201	} else if (executor instanceof StandardThreadExecutor) {
202	StandardThreadExecutor stdThreadExecutor = (StandardThreadExecutor) executor;
203	stdThreadExecutor.contextStopping();
204	}
205
206	}
207	}
208	}
209	}

Lines 16-27 Link Here

(-)java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java (-2 / +105 lines)
16	*/	16	*/
17	package org.apache.tomcat.util.threads;	17	package org.apache.tomcat.util.threads;
18		18
		19	import java.lang.Thread.UncaughtExceptionHandler;
19	import java.util.concurrent.BlockingQueue;	20	import java.util.concurrent.BlockingQueue;
20	import java.util.concurrent.RejectedExecutionException;	21	import java.util.concurrent.RejectedExecutionException;
21	import java.util.concurrent.RejectedExecutionHandler;	22	import java.util.concurrent.RejectedExecutionHandler;
22	import java.util.concurrent.ThreadFactory;	23	import java.util.concurrent.ThreadFactory;
23	import java.util.concurrent.TimeUnit;	24	import java.util.concurrent.TimeUnit;
24	import java.util.concurrent.atomic.AtomicInteger;	25	import java.util.concurrent.atomic.AtomicInteger;
		26	import java.util.concurrent.atomic.AtomicLong;
		27
		28	import org.apache.juli.logging.Log;
		29	import org.apache.juli.logging.LogFactory;
		30
25	/**	31	/**
26	* Same as a java.util.concurrent.ThreadPoolExecutor but implements a much more efficient	32	* Same as a java.util.concurrent.ThreadPoolExecutor but implements a much more efficient
27	* {@link #getSubmittedCount()} method, to be used to properly handle the work queue.	33	* {@link #getSubmittedCount()} method, to be used to properly handle the work queue.
Lines 31-37 Link Here
31	*	37	*
32	*/	38	*/
33	public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {	39	public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
34		40	private static final Log log = LogFactory.getLog(ThreadPoolExecutor.class);
		41
35	/**	42	/**
36	* The number of tasks submitted but not yet finished. This includes tasks	43	* The number of tasks submitted but not yet finished. This includes tasks
37	* in the queue and tasks that have been handed to a worker thread but the	44	* in the queue and tasks that have been handed to a worker thread but the
Lines 39-45 Link Here
39	* This number is always greater or equal to {@link #getActiveCount()}.	46	* This number is always greater or equal to {@link #getActiveCount()}.
40	*/	47	*/
41	private final AtomicInteger submittedCount = new AtomicInteger(0);	48	private final AtomicInteger submittedCount = new AtomicInteger(0);
42		49	private final AtomicLong lastContextStoppedTime = new AtomicLong(0L);
		50
		51	/**
		52	* Most recent time in ms when a thread decided to kill itself to avoid
		53	* potential memory leaks. Useful to throttle the rate of renewals of
		54	* threads.
		55	*/
		56	private final AtomicLong lastTimeThreadKilledItself = new AtomicLong(0L);
		57
		58	/**
		59	* Delay in ms between 2 threads being renewed. If negative, do not renew threads.
		60	*/
		61	private long threadRenewalDelay = 1000L;
		62
43	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {	63	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
44	super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);	64	super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
45	}	65	}
Lines 56-67 Link Here
56	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {	76	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
57	super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new RejectHandler());	77	super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new RejectHandler());
58	}	78	}
		79
		80	public long getThreadRenewalDelay() {
		81	return threadRenewalDelay;
		82	}
59		83
		84	public void setThreadRenewalDelay(long threadRenewalDelay) {
		85	this.threadRenewalDelay = threadRenewalDelay;
		86	}
		87
60	@Override	88	@Override
61	protected void afterExecute(Runnable r, Throwable t) {	89	protected void afterExecute(Runnable r, Throwable t) {
62	submittedCount.decrementAndGet();	90	submittedCount.decrementAndGet();
		91
		92	if (t == null) {
		93	stopCurrentThreadIfNeeded();
		94	}
63	}	95	}
64		96
		97	/**
		98	* If the current thread was started before the last time when a context was
		99	* stopped, an exception is thrown so that the current thread is stopped.
		100	*/
		101	protected void stopCurrentThreadIfNeeded() {
		102	if (currentThreadShouldBeStopped()) {
		103	long lastTime = lastTimeThreadKilledItself.longValue();
		104	if (lastTime + threadRenewalDelay < System.currentTimeMillis()) {
		105	if (lastTimeThreadKilledItself.compareAndSet(lastTime, System.currentTimeMillis() + 1)) {
		106	// OK, it's really time to dispose of this thread
		107
		108	final String msg = "Stopping thread " + Thread.currentThread().getName()
		109	+ " to avoid potential memory leaks after a context was stopped.";
		110	Thread.currentThread().setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
		111	@Override
		112	public void uncaughtException(Thread t, Throwable e) {
		113	// yes, swallow the exception
		114	log.debug(msg);
		115	}
		116	});
		117	throw new RuntimeException(msg);
		118	}
		119	}
		120	}
		121	}
		122
		123	protected boolean currentThreadShouldBeStopped() {
		124	if (threadRenewalDelay >= 0 && Thread.currentThread() instanceof TaskThread) {
		125	TaskThread currentTaskThread = (TaskThread) Thread.currentThread();
		126	if (currentTaskThread.getCreationTime() < this.lastContextStoppedTime.longValue()) {
		127	return true;
		128	}
		129	}
		130	return false;
		131	}
		132
65	public int getSubmittedCount() {	133	public int getSubmittedCount() {
66	return submittedCount.get();	134	return submittedCount.get();
67	}	135	}
Lines 111-116 Link Here
111		179
112	}	180	}
113	}	181	}
		182
		183	public void contextStopping() {
		184	this.lastContextStoppedTime.set(System.currentTimeMillis());
		185
		186	// save the current pool parameters to restore them later
		187	int savedCorePoolSize = this.getCorePoolSize();
		188	TaskQueue taskQueue = getQueue() instanceof TaskQueue ? (TaskQueue) getQueue() : null;
		189	if (taskQueue != null) {
		190	// note by slaurent : quite oddly threadPoolExecutor.setCorePoolSize
		191	// checks that queue.remainingCapacity()==0. I did not understand
		192	// why, but to get the intended effect of waking up idle threads, I
		193	// temporarily fake this condition.
		194	taskQueue.setForcedRemainingCapacity(0);
		195	}
		196
		197	// setCorePoolSize(0) wakes idle threads
		198	this.setCorePoolSize(0);
		199
		200	// wait a little so that idle threads wake and poll the queue again,
		201	// this time always with a timeout (queue.poll() instead of queue.take())
		202	// even if we did not wait enough, TaskQueue.take() takes care of timing out
		203	// so that we are sure that all threads of the pool are renewed in a limited
		204	// time, something like (threadKeepAlive + longest request time)
		205	try {
		206	Thread.sleep(200L);
		207	} catch (InterruptedException e) {
		208	//yes, ignore
		209	}
		210
		211	if (taskQueue != null) {
		212	// ok, restore the state of the queue and pool
		213	taskQueue.setForcedRemainingCapacity(null);
		214	}
		215	this.setCorePoolSize(savedCorePoolSize);
		216	}
114		217
115	private static class RejectHandler implements RejectedExecutionHandler {	218	private static class RejectHandler implements RejectedExecutionHandler {
116	@Override	219	@Override