Attachment #23005 for bug #46284

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

Lines 36-41 Link Here

(-)java/org/apache/catalina/Globals.java (+8 lines)
36	"org.apache.catalina.deploy.alt_dd";	36	"org.apache.catalina.deploy.alt_dd";
37		37
38	/**	38	/**
		39	* The servlet context attribute under which we store the concurrent latch
		40	* used to block processing cluster messages until after local application
		41	* initialization
		42	*/
		43	public static final String CLUSTER_DELAY =
		44	"org.apache.catalina.ha.delay";
		45
		46	/**
39	* The request attribute under which we store the array of X509Certificate	47	* The request attribute under which we store the array of X509Certificate
40	* objects representing the certificate chain presented by our client,	48	* objects representing the certificate chain presented by our client,
41	* if any.	49	* if any.




import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;

import javax.servlet.ServletContext;

import org.apache.catalina.Cluster;
import org.apache.catalina.Container;
import org.apache.catalina.Context;
import org.apache.catalina.Engine;
import org.apache.catalina.Globals;
import org.apache.catalina.Host;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.LifecycleListener;

import org.apache.catalina.Valve;
import org.apache.catalina.core.StandardContext;
import org.apache.catalina.ha.CatalinaCluster;
import org.apache.catalina.ha.ClusterManager;
import org.apache.catalina.ha.ClusterMessage;
import org.apache.catalina.ha.tcp.ReplicationValve;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.io.ReplicationStream;
import org.apache.catalina.util.LifecycleSupport;
import org.apache.catalina.util.StringManager;
import org.apache.catalina.ha.ClusterManager;

/**
 * The DeltaManager manages replicated sessions by only replicating the deltas

 * @author Craig R. McClanahan
 * @author Jean-Francois Arcand
 * @author Peter Rossbach
 * @author Jason Lunn
 * @version $Revision$ $Date$
 */

public class DeltaManager extends ClusterManagerBase {

    // ---------------------------------------------------- Security Classes
    public static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(DeltaManager.class);

    protected LifecycleSupport lifecycle = new LifecycleSupport(this);

    /**
     * Flag indicating that messageReceived(ClusterMessage) should block until
     * all local applications have completed initialization
     */
    protected boolean delay = false;
    
    /**
     * Barrier used to receive notification from other threads that it is okay
     * to process incoming messages from the cluster
     */
    private volatile CountDownLatch gate = null;
    
    /**
     * The maximum number of active Sessions allowed, or -1 for no limit.
     */
    private int maxActiveSessions = -1;

    /**
     * wait time between send session block (default 2 sec) 
     */
    private int sendAllSessionsWaitTime = 2 * 1000; 
    private ArrayList<SessionMessage> receivedMessageQueue = new ArrayList<SessionMessage>() ;
    private boolean receiverQueue = false ;
    private boolean stateTimestampDrop = true ;
    private long stateTransferCreateSendTime; 

    public String getName() {
        return name;
    }
    
    /**
     * Set the member that indicates processing messages should wait for local
     * initialization of applications to complete
     * 
     * @param delayed If true, processing messageReceived(ClusterMessage) will
     * block until local web applications have completed initialization.
     */
    public void setDelay ( boolean delay ) {
        this.delay = delay;
    }
    
    /**
     * Gets the member the delayed flag
     * @return delayed - boolean flag indicating that
     * messageReceived(ClusterMessage) should block until local initialization
     * of applications has completed
     */
    public boolean getDelay () {
        return delay;
    }

    /**
     * @return Returns the counterSend_EVT_GET_ALL_SESSIONS.

        lifecycle.removeLifecycleListener(listener);
    }

    @Override
    /**
     * If this.delay is true, create a gate that will be opened when
     * local application initialization is complete  
     */
    public void init () {
        if (delay && container != null) {
            if (container instanceof StandardContext) {
                ServletContext servletContext = ((StandardContext) container).getServletContext();
                if (servletContext != null) {
                    gate = new CountDownLatch(1);
                    servletContext.setAttribute(Globals.CLUSTER_DELAY, gate);
                }
            }
        }
        super.init();
    }

    /**
     * Prepare for the beginning of active use of the public methods of this
     * component. This method should be called after <code>configure()</code>,

                waitForSendAllSessions(beforeSendTime);
            } finally {
                synchronized(receivedMessageQueue) {
                    for (Iterator<SessionMessage> iter = receivedMessageQueue.iterator(); iter.hasNext();) {
                        SessionMessage smsg = iter.next();
                        if (!stateTimestampDrop) {
                            messageReceived(smsg, smsg.getAddress() != null ? (Member) smsg.getAddress() : null);
                        } else {

     *            the message received.
     */
    public void messageDataReceived(ClusterMessage cmsg) {
        // Block processing until local application initialization has
        // completed, if a gate has been erected
        synchronized (this) {
            if(gate != null) {
                try {
                    gate.await();
                    gate = null;
                }
                catch(InterruptedException e) {
                    log.error(e, e);
                }
            }
        }

        if (cmsg != null && cmsg instanceof SessionMessage) {
            SessionMessage msg = (SessionMessage) cmsg;
            switch (msg.getEventType()) {

        result.sendAllSessionsWaitTime = sendAllSessionsWaitTime ; 
        result.receiverQueue = receiverQueue ;
        result.stateTimestampDrop = stateTimestampDrop ;
        result.stateTransferCreateSendTime = stateTransferCreateSendTime;
        result.delay = delay;
        return result;
    }
}

Lines 268-274 Link Here

(-)java/org/apache/catalina/ha/session/mbeans-descriptors.xml (-1 / +5 lines)
268	<attribute	268	<attribute
269	name="expireSessionsOnShutdown"	269	name="expireSessionsOnShutdown"
270	is="true"	270	is="true"
271	description="exipre all sessions cluster wide as one node goes down"	271	description="expire all sessions cluster wide as one node goes down"
272	type="boolean"/>	272	type="boolean"/>
273	<attribute	273	<attribute
274	name="notifyListenersOnReplication"	274	name="notifyListenersOnReplication"
Lines 293-298 Link Here
293	name="sendAllSessionsWaitTime"	293	name="sendAllSessionsWaitTime"
294	description="wait time between send session block (default 2 sec)"	294	description="wait time between send session block (default 2 sec)"
295	type="int"/>	295	type="int"/>
		296	<attribute
		297	name="delay"
		298	description="wait until local applications have initialized before processing cluster messages"
		299	type="boolean"/>
296	<operation	300	<operation
297	name="listSessionIds"	301	name="listSessionIds"
298	description="Return the list of active session ids"	302	description="Return the list of active session ids"

Lines 28-38 Link Here

(-)java/org/apache/catalina/startup/Catalina.java (-10 / +57 lines)
28	import java.util.ArrayList;	28	import java.util.ArrayList;
29	import java.util.HashMap;	29	import java.util.HashMap;
30	import java.util.List;	30	import java.util.List;
		31	import java.util.concurrent.CountDownLatch;
		32
		33	import javax.servlet.ServletContext;
31		34
32	import org.apache.catalina.Container;	35	import org.apache.catalina.Container;
		36	import org.apache.catalina.Globals;
33	import org.apache.catalina.Lifecycle;	37	import org.apache.catalina.Lifecycle;
34	import org.apache.catalina.LifecycleException;	38	import org.apache.catalina.LifecycleException;
35	import org.apache.catalina.Server;	39	import org.apache.catalina.Server;
		40	import org.apache.catalina.Service;
		41	import org.apache.catalina.core.ContainerBase;
		42	import org.apache.catalina.core.StandardContext;
36	import org.apache.catalina.core.StandardServer;	43	import org.apache.catalina.core.StandardServer;
37	import org.apache.tomcat.util.digester.Digester;	44	import org.apache.tomcat.util.digester.Digester;
38	import org.apache.tomcat.util.digester.Rule;	45	import org.apache.tomcat.util.digester.Rule;
Lines 56-61 Link Here
56	*	63	*
57	* @author Craig R. McClanahan	64	* @author Craig R. McClanahan
58	* @author Remy Maucherat	65	* @author Remy Maucherat
		66	* @author Jason Lunn
59	* @version $Revision$ $Date$	67	* @version $Revision$ $Date$
60	*/	68	*/
61		69
Lines 579-585 Link Here
579	log.error("Catalina.start: ", e);	587	log.error("Catalina.start: ", e);
580	}	588	}
581	}	589	}
582		590
		591	// Open any gates stored in ServletContexts to allow the processing of
		592	// cluster messages once local applications have been initialized
		593	Service [] services = server.findServices();
		594	if ( services != null ) {
		595	for ( Service service : services ) {
		596	if ( service != null ) {
		597	openContainerGates( service.getContainer() );
		598	}
		599	}
		600	}
		601
583	long t2 = System.nanoTime();	602	long t2 = System.nanoTime();
584	if(log.isInfoEnabled())	603	if(log.isInfoEnabled())
585	log.info("Server startup in " + ((t2 - t1) / 1000000) + " ms");	604	log.info("Server startup in " + ((t2 - t1) / 1000000) + " ms");
Lines 601-607 Link Here
601	await();	620	await();
602	stop();	621	stop();
603	}	622	}
604
605	}	623	}
606		624
607		625
Lines 609-615 Link Here
609	* Stop an existing server instance.	627	* Stop an existing server instance.
610	*/	628	*/
611	public void stop() {	629	public void stop() {
612
613	try {	630	try {
614	// Remove the ShutdownHook first so that server.stop()	631	// Remove the ShutdownHook first so that server.stop()
615	// doesn't get invoked twice	632	// doesn't get invoked twice
Lines 629-635 Link Here
629	log.error("Catalina.stop", e);	646	log.error("Catalina.stop", e);
630	}	647	}
631	}	648	}
632
633	}	649	}
634		650
635		651
Lines 637-645 Link Here
637	* Await and shutdown.	653	* Await and shutdown.
638	*/	654	*/
639	public void await() {	655	public void await() {
640
641	server.await();	656	server.await();
642
643	}	657	}
644		658
645		659
Lines 647-661 Link Here
647	* Print usage information for this application.	661	* Print usage information for this application.
648	*/	662	*/
649	protected void usage() {	663	protected void usage() {
650
651	System.out.println	664	System.out.println
652	("usage: java org.apache.catalina.startup.Catalina"	665	("usage: java org.apache.catalina.startup.Catalina"
653	+ " [ -config {pathname} ]"	666	+ " [ -config {pathname} ]"
654	+ " [ -nonaming ] { start \| stop }");	667	+ " [ -nonaming ] { start \| stop }");
655
656	}	668	}
657		669
658		670
		671	/**
		672	* Traverses the argument container and decrements the CountDownLatch
		673	* found in the servlet context with attribute key Globals.CLUSTER_DELAY
		674	* if found
		675	* @param container Possibly null Container instance
		676	*/
		677	protected void openContainerGates ( Container container ) {
		678	if (container == null) {
		679	return;
		680	}
		681
		682	if ( container instanceof StandardContext ) {
		683	StandardContext context =
		684	(StandardContext)container;
		685	ServletContext servletContext =
		686	context.getServletContext();
		687	if (servletContext != null) {
		688	Object contextAttribute = servletContext
		689	.getAttribute(Globals.CLUSTER_DELAY);
		690	if (contextAttribute != null &&
		691	contextAttribute instanceof CountDownLatch) {
		692	CountDownLatch gate =
		693	(CountDownLatch) contextAttribute;
		694	gate.countDown();
		695	}
		696	}
		697	} else if ( container instanceof ContainerBase ) {
		698	ContainerBase base = (ContainerBase)container;
		699	Container [] containers = base.findChildren();
		700	if ( containers != null ) {
		701	for ( Container childContainer : containers ) {
		702	openContainerGates( childContainer );
		703	}
		704	}
		705	}
		706	}
		707
659	// --------------------------------------- CatalinaShutdownHook Inner Class	708	// --------------------------------------- CatalinaShutdownHook Inner Class
660		709
661	// XXX Should be moved to embedded !	710	// XXX Should be moved to embedded !
Lines 709-714 Link Here
709	top.setParentClassLoader(parentClassLoader);	758	top.setParentClassLoader(parentClassLoader);
710		759
711	}	760	}
712
713
714	}	761	}

Return to bug 46284

Lines 26-36 Link Here

(-)java/org/apache/catalina/ha/session/DeltaManager.java (-7 / +78 lines)
26	import java.util.ArrayList;	26	import java.util.ArrayList;
27	import java.util.Date;	27	import java.util.Date;
28	import java.util.Iterator;	28	import java.util.Iterator;
		29	import java.util.concurrent.CountDownLatch;
		30
		31	import javax.servlet.ServletContext;
29		32
30	import org.apache.catalina.Cluster;	33	import org.apache.catalina.Cluster;
31	import org.apache.catalina.Container;	34	import org.apache.catalina.Container;
32	import org.apache.catalina.Context;	35	import org.apache.catalina.Context;
33	import org.apache.catalina.Engine;	36	import org.apache.catalina.Engine;
		37	import org.apache.catalina.Globals;
34	import org.apache.catalina.Host;	38	import org.apache.catalina.Host;
35	import org.apache.catalina.LifecycleException;	39	import org.apache.catalina.LifecycleException;
36	import org.apache.catalina.LifecycleListener;	40	import org.apache.catalina.LifecycleListener;
Lines 38-50 Link Here
38	import org.apache.catalina.Valve;	42	import org.apache.catalina.Valve;
39	import org.apache.catalina.core.StandardContext;	43	import org.apache.catalina.core.StandardContext;
40	import org.apache.catalina.ha.CatalinaCluster;	44	import org.apache.catalina.ha.CatalinaCluster;
		45	import org.apache.catalina.ha.ClusterManager;
41	import org.apache.catalina.ha.ClusterMessage;	46	import org.apache.catalina.ha.ClusterMessage;
42	import org.apache.catalina.ha.tcp.ReplicationValve;	47	import org.apache.catalina.ha.tcp.ReplicationValve;
43	import org.apache.catalina.tribes.Member;	48	import org.apache.catalina.tribes.Member;
44	import org.apache.catalina.tribes.io.ReplicationStream;	49	import org.apache.catalina.tribes.io.ReplicationStream;
45	import org.apache.catalina.util.LifecycleSupport;	50	import org.apache.catalina.util.LifecycleSupport;
46	import org.apache.catalina.util.StringManager;	51	import org.apache.catalina.util.StringManager;
47	import org.apache.catalina.ha.ClusterManager;
48		52
49	/**	53	/**
50	* The DeltaManager manages replicated sessions by only replicating the deltas	54	* The DeltaManager manages replicated sessions by only replicating the deltas
Lines 62-71 Link Here
62	* @author Craig R. McClanahan	66	* @author Craig R. McClanahan
63	* @author Jean-Francois Arcand	67	* @author Jean-Francois Arcand
64	* @author Peter Rossbach	68	* @author Peter Rossbach
		69	* @author Jason Lunn
65	* @version $Revision$ $Date$	70	* @version $Revision$ $Date$
66	*/	71	*/
67		72
68	public class DeltaManager extends ClusterManagerBase{	73	public class DeltaManager extends ClusterManagerBase {
69		74
70	// ---------------------------------------------------- Security Classes	75	// ---------------------------------------------------- Security Classes
71	public static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(DeltaManager.class);	76	public static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(DeltaManager.class);
Lines 106-111 Link Here
106	protected LifecycleSupport lifecycle = new LifecycleSupport(this);	111	protected LifecycleSupport lifecycle = new LifecycleSupport(this);
107		112
108	/**	113	/**
		114	* Flag indicating that messageReceived(ClusterMessage) should block until
		115	* all local applications have completed initialization
		116	*/
		117	protected boolean delay = false;
		118
		119	/**
		120	* Barrier used to receive notification from other threads that it is okay
		121	* to process incoming messages from the cluster
		122	*/
		123	private volatile CountDownLatch gate = null;
		124
		125	/**
109	* The maximum number of active Sessions allowed, or -1 for no limit.	126	* The maximum number of active Sessions allowed, or -1 for no limit.
110	*/	127	*/
111	private int maxActiveSessions = -1;	128	private int maxActiveSessions = -1;
Lines 121-128 Link Here
121	/**	138	/**
122	* wait time between send session block (default 2 sec)	139	* wait time between send session block (default 2 sec)
123	*/	140	*/
124	private int sendAllSessionsWaitTime = 2 * 1000 ;	141	private int sendAllSessionsWaitTime = 2 * 1000;
125	private ArrayList receivedMessageQueue = new ArrayList() ;	142	private ArrayList<SessionMessage> receivedMessageQueue = new ArrayList<SessionMessage>() ;
126	private boolean receiverQueue = false ;	143	private boolean receiverQueue = false ;
127	private boolean stateTimestampDrop = true ;	144	private boolean stateTimestampDrop = true ;
128	private long stateTransferCreateSendTime;	145	private long stateTransferCreateSendTime;
Lines 175-180 Link Here
175	public String getName() {	192	public String getName() {
176	return name;	193	return name;
177	}	194	}
		195
		196	/**
		197	* Set the member that indicates processing messages should wait for local
		198	* initialization of applications to complete
		199	*
		200	* @param delayed If true, processing messageReceived(ClusterMessage) will
		201	* block until local web applications have completed initialization.
		202	*/
		203	public void setDelay ( boolean delay ) {
		204	this.delay = delay;
		205	}
		206
		207	/**
		208	* Gets the member the delayed flag
		209	* @return delayed - boolean flag indicating that
		210	* messageReceived(ClusterMessage) should block until local initialization
		211	* of applications has completed
		212	*/
		213	public boolean getDelay () {
		214	return delay;
		215	}
178		216
179	/**	217	/**
180	* @return Returns the counterSend_EVT_GET_ALL_SESSIONS.	218	* @return Returns the counterSend_EVT_GET_ALL_SESSIONS.
Lines 781-786 Link Here
781	lifecycle.removeLifecycleListener(listener);	819	lifecycle.removeLifecycleListener(listener);
782	}	820	}
783		821
		822	@Override
		823	/**
		824	* If this.delay is true, create a gate that will be opened when
		825	* local application initialization is complete
		826	*/
		827	public void init () {
		828	if (delay && container != null) {
		829	if (container instanceof StandardContext) {
		830	ServletContext servletContext = ((StandardContext) container).getServletContext();
		831	if (servletContext != null) {
		832	gate = new CountDownLatch(1);
		833	servletContext.setAttribute(Globals.CLUSTER_DELAY, gate);
		834	}
		835	}
		836	}
		837	super.init();
		838	}
		839
784	/**	840	/**
785	* Prepare for the beginning of active use of the public methods of this	841	* Prepare for the beginning of active use of the public methods of this
786	* component. This method should be called after <code>configure()</code>,	842	* component. This method should be called after <code>configure()</code>,
Lines 885-892 Link Here
885	waitForSendAllSessions(beforeSendTime);	941	waitForSendAllSessions(beforeSendTime);
886	} finally {	942	} finally {
887	synchronized(receivedMessageQueue) {	943	synchronized(receivedMessageQueue) {
888	for (Iterator iter = receivedMessageQueue.iterator(); iter.hasNext();) {	944	for (Iterator<SessionMessage> iter = receivedMessageQueue.iterator(); iter.hasNext();) {
889	SessionMessage smsg = (SessionMessage) iter.next();	945	SessionMessage smsg = iter.next();
890	if (!stateTimestampDrop) {	946	if (!stateTimestampDrop) {
891	messageReceived(smsg, smsg.getAddress() != null ? (Member) smsg.getAddress() : null);	947	messageReceived(smsg, smsg.getAddress() != null ? (Member) smsg.getAddress() : null);
892	} else {	948	} else {
Lines 1068-1073 Link Here
1068	* the message received.	1124	* the message received.
1069	*/	1125	*/
1070	public void messageDataReceived(ClusterMessage cmsg) {	1126	public void messageDataReceived(ClusterMessage cmsg) {
		1127	// Block processing until local application initialization has
		1128	// completed, if a gate has been erected
		1129	synchronized (this) {
		1130	if(gate != null) {
		1131	try {
		1132	gate.await();
		1133	gate = null;
		1134	}
		1135	catch(InterruptedException e) {
		1136	log.error(e, e);
		1137	}
		1138	}
		1139	}
		1140
1071	if (cmsg != null && cmsg instanceof SessionMessage) {	1141	if (cmsg != null && cmsg instanceof SessionMessage) {
1072	SessionMessage msg = (SessionMessage) cmsg;	1142	SessionMessage msg = (SessionMessage) cmsg;
1073	switch (msg.getEventType()) {	1143	switch (msg.getEventType()) {
Lines 1535-1541 Link Here
1535	result.sendAllSessionsWaitTime = sendAllSessionsWaitTime ;	1605	result.sendAllSessionsWaitTime = sendAllSessionsWaitTime ;
1536	result.receiverQueue = receiverQueue ;	1606	result.receiverQueue = receiverQueue ;
1537	result.stateTimestampDrop = stateTimestampDrop ;	1607	result.stateTimestampDrop = stateTimestampDrop ;
1538	result.stateTransferCreateSendTime = stateTransferCreateSendTime;	1608	result.stateTransferCreateSendTime = stateTransferCreateSendTime;
		1609	result.delay = delay;
1539	return result;	1610	return result;
1540	}	1611	}
1541	}	1612	}