Attachment #195673 for bug #337037

This Bugzilla instance is deprecated, and most Eclipse projects now use GitHub or Eclipse GitLab. Please see the deprecation plan for details.

Lines 5-11 Link Here

(-)examples/org.eclipse.persistence.example.distributed.CollatzEJB/ejbModule/META-INF/persistence.xml (-1 / +1 lines)
5	<provider>org.eclipse.persistence.jpa.PersistenceProvider</provider>	5	<provider>org.eclipse.persistence.jpa.PersistenceProvider</provider>
6	<!-- provider>org.hibernate.ejb.HibernatePersistence</provider-->	6	<!-- provider>org.hibernate.ejb.HibernatePersistence</provider-->
7	<!-- jta-data-source>jdbc/Collatz2</jta-data-source--> <!-- glassfish -->	7	<!-- jta-data-source>jdbc/Collatz2</jta-data-source--> <!-- glassfish -->
8	<jta-data-source>collatzRemote</jta-data-source> <!-- weblogic -->	8	<jta-data-source>collatzRemote2</jta-data-source> <!-- weblogic -->
9	<!-- mapping-file>META-INF/orm.xml</mapping-file-->	9	<!-- mapping-file>META-INF/orm.xml</mapping-file-->
10	<jar-file>org.eclipse.persistence.example.distributed.CollatzModel.jar</jar-file>	10	<jar-file>org.eclipse.persistence.example.distributed.CollatzModel.jar</jar-file>
11	<class>org.eclipse.persistence.example.distributed.collatz.model.ActiveProcessor</class>	11	<class>org.eclipse.persistence.example.distributed.collatz.model.ActiveProcessor</class>

Lines 146-151 Link Here

(-)examples/org.eclipse.persistence.example.distributed.CollatzModel/src/org/eclipse/persistence/example/distributed/collatz/model/UnitOfWork.java (+2 lines)
146	buffer = new StringBuffer("_collatz: ");	146	buffer = new StringBuffer("_collatz: ");
147	buffer.append(System.currentTimeMillis());	147	buffer.append(System.currentTimeMillis());
148	buffer.append(": ");	148	buffer.append(": ");
		149	buffer.append(getProcessor().getIdentifier());
		150	buffer.append(": ");
149	buffer.append(prefix);	151	buffer.append(prefix);
150	buffer.append(",");	152	buffer.append(",");
151	buffer.append(this.getInterval());	153	buffer.append(this.getInterval());




 *     16/02/2011 2.3  Michael O'Brien 
 *          - 337037: initial API and implementation platform to be used for 
 *             distributed EE application research, development and architecture
 *     16/02/2011 2.3  Add multithreaded capability (default is hard + soft (HT) cores = ~90% multicore CPU utilization)  
 ******************************************************************************/  
package org.eclipse.persistence.example.distributed.collatz.presentation;


import java.util.Map;

//import javax.ejb.EJBException;
import javax.naming.CommunicationException;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NameNotFoundException;


import org.eclipse.persistence.example.distributed.collatz.business.CollatzFacadeRemote;
import org.eclipse.persistence.example.distributed.collatz.model.UnitOfWork;
//import weblogic.utils.net.SocketResetException;
//import javax.persistence.OptimisticLockException;

/**

     * Numbers greater than this are encountered in scientific, cryptographic and nanosecond time sensitive calculations. 
     */
    private static final Long MAX_BIGINTEGER_IN_SQL = Long.MAX_VALUE;
    /** Get number of (hyperthreaded + real) cores.  IE: p630 with HT=2, Core2 E8400=2 and Core i7-920 = 8 */
    public static final int CORES = Runtime.getRuntime().availableProcessors() << 0;

    // TODO: we need to move all this to a properties file
    /** this is the index of the current server - for use in the maps and lists below */

    /** output cached context maps for each remote server */
    private Map<String, Context> rmiCachedContextMap = new HashMap<String, Context>();
    /** $Proxy remote objects */
    private List<Map<String, CollatzFacadeRemote>> remoteObjects = new ArrayList<Map<String, CollatzFacadeRemote>>(CORES);
    //private Map<String, CollatzFacadeRemote> remoteObjects = new HashMap<String, CollatzFacadeRemote>();
    /** How many processors are available (real + hyperthreaded) */
    private Map<String, Integer> availableProcessors = new HashMap<String, Integer>();
    /** whether the node is accepting requests or not */

    /** map of t3 protocol URLs */
    private Map<String, String>  serverIPMap = new HashMap<String, String>();

    //private List<CollatzRunnable> runnables = new ArrayList<CollatzRunnable>();
    
    private static final String DEFAULT_CLIENT_NAME = "default";
    // WebLogic
    // verify that all EE libraries available via http://download.oracle.com/docs/cd/E12840_01/wls/docs103/client/jarbuilder.html

            aTable.put(Context.INITIAL_CONTEXT_FACTORY,CONTEXT_FACTORY_NAME);
            aTable.put(Context.PROVIDER_URL, serverT3[i]);
        }
        for(int i=0;i<CORES;i++) {
            connect(i);
        }
    }
    
    public void connect(int threadID) {
        // Setup RMI Objects
        // Establish RMI connections to the session beans
        //for(String aServer : serverNames) {
        String aServer = serverNames.get(0);
            Context aContext = null;
            try {
                // no need to set the host if on same machine
                aContext = new InitialContext(contextHashtableMap.get(aServer));
                rmiCachedContextMap.put(aServer, aContext);
                System.out.println("_collatz: " + System.currentTimeMillis() + ": Context for " + (aServer + threadID) + " : " + aContext);
                // For qualified name look for weblogic log "EJB Deployed EJB with JNDI name"
                Object aRemoteReference = aContext.lookup(SESSION_BEAN_REMOTE_NAME);
                System.out.println("_collatz: " + System.currentTimeMillis() + ": Remote Object: " + aRemoteReference);
                // narrow the $proxy remote bean
                CollatzFacadeRemote aNode = (CollatzFacadeRemote) PortableRemoteObject.narrow(aRemoteReference, CollatzFacadeRemote.class);
                Map<String, CollatzFacadeRemote> remoteObject = new HashMap<String, CollatzFacadeRemote>();
                remoteObject.put(aServer, aNode);
                remoteObjects.add(remoteObject);
            } catch (Exception ce) {
                // server down throws a javax.naming.CommunicationException inside a java.net.ConnectException
                ce.printStackTrace();
                // mark the current node as down, clear the flag in 5 min
            }
        //}
    }

    private CollatzFacadeRemote lookupRemoteBean(String aServer, int threadID)  {
        CollatzFacadeRemote remoteBean = null;
        try {
            Context aContext = rmiCachedContextMap.get(aServer);

            boolean remoteLookupSuccess = false;
            int lookupIterations = 0;
            while(!remoteLookupSuccess && aRemoteReference == null && lookupIterations < 50) {
                System.out.println("_collatz: " + System.currentTimeMillis() + ": Thread: " + threadID + " : Context lookup for " + SESSION_BEAN_REMOTE_NAME + " from: " + aContext);
                try {
                    aRemoteReference = aContext.lookup(SESSION_BEAN_REMOTE_NAME);
                } catch (NameNotFoundException nnfe) {
                    System.out.println(nnfe.getMessage());
                    System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - possible redeploy in progress on central server: " + lookupIterations);
                    Thread.sleep(1000);
                } catch (CommunicationException ce) {//SocketResetException sre) {
                    // Network was temporarily disconnected - or server went down
                    System.out.println(ce.getMessage());
                    System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - Network or server is temporarily down: " + lookupIterations);
                    Thread.sleep(1000);
                }
                lookupIterations++;
            }
            System.out.println("_collatz: " + System.currentTimeMillis() + ": Remote Object: " + aRemoteReference);
            // narrow the $proxy remote bean
            remoteBean = (CollatzFacadeRemote) PortableRemoteObject.narrow(aRemoteReference, CollatzFacadeRemote.class);
            Map<String, CollatzFacadeRemote> remoteObject = new HashMap<String, CollatzFacadeRemote>();
            remoteObject.put(aServer, remoteBean);
            setRemoteObjects(threadID, remoteObject);
        /*} catch (ConnectException rmice) {//CommunicationException ce) {//SocketResetException sre) {
            // Network was temporarily disconnected - or server went down
            System.out.println(rmice.getMessage());
            System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - Network or server is temporarily down: " + lookupIterations);
            Thread.sleep(1000);*/
        } catch (Exception e) {
            e.printStackTrace();
        }
        return remoteBean;
     }
    
    public void processUnitOfWork(int threadID) {
        // ask for a work packet
        UnitOfWork uow = null;
        CollatzFacadeRemote collatzFacade = null;
        StringBuffer aBuffer = new StringBuffer();
        String threadName;
        // Endlessly generate RMI requests
        for(;;) {
            try {
                // Send messages to entire grid in parallel if we connect to more than one server
                // TODO: create Threads for each remoteObject
                for(String remoteServer : remoteObjects.get(threadID).keySet()) {
                    threadName = serverDNS[serverInUse] + threadID;
                    collatzFacade = remoteObjects.get(threadID).get(remoteServer);
                    try {
                        // Issue: One JVM halt will affect the entire distributed app.
                        // don't let a node failure halt the host
                        // this remote call can throw an EJBException wrapping a java.rmi.ConnectException                            
                        uow = collatzFacade.requestUnitOfWork(threadName,availableProcessors.get(remoteServer)); 
                        if(null == uow) {
                            // possible redeploy
                            System.out.println("_collatz: " + System.currentTimeMillis() + ": persistence not functioning on server " + serverDNS[serverInUse]);
                        } else {
                            aBuffer = new StringBuffer("_collatz: ");
                            aBuffer.append(System.currentTimeMillis());
                            aBuffer.append(": processing UnitOfWork: ");
                            aBuffer.append(uow);
                            aBuffer.append(" ID#");
                            aBuffer.append(uow.getId());
                            aBuffer.append(" ");
                            aBuffer.append(uow.getInitial());
                            aBuffer.append("-");
                            aBuffer.append(uow.getExtent());
                            aBuffer.append(" for: ");
                            aBuffer.append(threadName);
                            System.out.println(aBuffer.toString());
                        }
                    } catch (Exception e) {//(EJBException e) {
                        //  weblogic.transaction.internal.TimedOutException: Transaction timed out after 29 seconds
                        // or SQLException on constraint violation
                        // EJBException wrapping a java.rmi.ConnectException if the server is not running
                        e.printStackTrace();
                        // mark the current node as down, clear the flag in 5 min
                        //nodeUnavailable.put(remoteServer, true);
                    }
                    // compute collatz for the sequence
                    uow.processInterval();
                    Thread.yield(); // 
                    // return the results to the server
                    // don't cache the remote bean (it may be GC'd or redeployed)
                    collatzFacade = lookupRemoteBean(remoteServer, threadID);
                    boolean retry = true;
                    while(retry) {
                        try {
                            collatzFacade.postUnitOfWork(uow,retry); // possible EJBException or OptimisticLockException
                            retry = false;
                        } catch (Exception ole) {//OptimisticLockException ole) {
                            //System.out.println(ole.getMessage());
                            retry = true;
                            Thread.sleep(1000);
                        }
                    }
                    aBuffer = new StringBuffer("_collatz: ");
                    aBuffer.append(System.currentTimeMillis());
                    aBuffer.append(": results sent to server after ");
                    aBuffer.append(uow.getEndTimestamp() - uow.getStartTimestamp());
                    aBuffer.append(" ms @ ");
                    aBuffer.append(uow.getMIPS());
                    aBuffer.append(" MIPS");
                    System.out.println(aBuffer.toString());
                }
            } catch (Exception e) {
                e.printStackTrace();
                try {
                    Thread.sleep(10000);
                } catch (Exception ex) { }
            }
        }
    }

    protected void startThreads(int numberOfThreads) {
        threadSafetyPrivate(numberOfThreads);
    }
    
    protected void threadSafetyPrivate(int numberOfThreads) {
        List<Thread> threadList = new ArrayList<Thread>();
        for(int i=0; i<numberOfThreads; i++) {
            Thread aThread = new Thread(new CollatzRunnable(i));
            threadList.add(aThread);
            // stagger the threads so they are not in lockstep
            try {
                Thread.sleep(3000);
            } catch (Exception e) {           }
            aThread.start();
        }

        // Wait for [threadNumber] threads to complete before ending 
        for(Thread aThread : threadList) {
            try {
                synchronized (aThread) {
                    aThread.join();
                }
            } catch (InterruptedException ie_Ignored) {
                ie_Ignored.printStackTrace();
            } // The InterruptedException can be ignored 
        }
    }
    
    // Inner class implements Runnable instead of extending Thread directly
    class CollatzRunnable implements Runnable {
        protected int id;
        
        public CollatzRunnable(int anId) {
            id = anId;
        }
        
        public void run() {
            //connect(id);
            // We loop an arbitrary number of iterations inside each thread
            processUnitOfWork(id);
        }
    }
    
    /**
     * @param args the command line arguments
     */

        serverT3[serverInUse] = ip;

        SEClient client = new SEClient();
        // Create and start threads
        client.startThreads(CORES);
        //client.connect();
        //client.processUnitOfWork();
    }

    public Map<String, Hashtable<String, String>> getContextMap() {        return contextHashtableMap;    }
    public void setContextMap(Map<String, Hashtable<String, String>> contextMap) {        this.contextHashtableMap = contextMap;    }
    public Map<String, Context> getRmiContextMap() {        return rmiCachedContextMap;    }
    public void setRmiContextMap(Map<String, Context> rmiContextMap) {        this.rmiCachedContextMap = rmiContextMap;    }
    public Map<String, CollatzFacadeRemote> getRemoteObjects(int threadID) {        return remoteObjects.get(threadID);    }
    public void setRemoteObjects(int threadID, Map<String, CollatzFacadeRemote> remoteObjects) {        this.remoteObjects.set(threadID, remoteObjects);    }
    public Map<String, Boolean> getNodeUnavailable() {        return nodeUnavailable;    }
    public void setNodeUnavailable(Map<String, Boolean> nodeUnavailable) {        this.nodeUnavailable = nodeUnavailable;    }
    public Map<String, String> getServerIPMap() {        return serverIPMap;    }

Return to bug 337037

Lines 11-16 Link Here

(-)examples/org.eclipse.persistence.example.distributed.CollatzSE/src/org/eclipse/persistence/example/distributed/collatz/presentation/SEClient.java (-77 / +152 lines)
11	* 16/02/2011 2.3 Michael O'Brien	11	* 16/02/2011 2.3 Michael O'Brien
12	* - 337037: initial API and implementation platform to be used for	12	* - 337037: initial API and implementation platform to be used for
13	* distributed EE application research, development and architecture	13	* distributed EE application research, development and architecture
		14	* 16/02/2011 2.3 Add multithreaded capability (default is hard + soft (HT) cores = ~90% multicore CPU utilization)
14	******************************************************************************/	15	******************************************************************************/
15	package org.eclipse.persistence.example.distributed.collatz.presentation;	16	package org.eclipse.persistence.example.distributed.collatz.presentation;
16		17
Lines 21-26 Link Here
21	import java.util.Map;	22	import java.util.Map;
22		23
23	//import javax.ejb.EJBException;	24	//import javax.ejb.EJBException;
		25	import javax.naming.CommunicationException;
24	import javax.naming.Context;	26	import javax.naming.Context;
25	import javax.naming.InitialContext;	27	import javax.naming.InitialContext;
26	import javax.naming.NameNotFoundException;	28	import javax.naming.NameNotFoundException;
Lines 28-33 Link Here
28		30
29	import org.eclipse.persistence.example.distributed.collatz.business.CollatzFacadeRemote;	31	import org.eclipse.persistence.example.distributed.collatz.business.CollatzFacadeRemote;
30	import org.eclipse.persistence.example.distributed.collatz.model.UnitOfWork;	32	import org.eclipse.persistence.example.distributed.collatz.model.UnitOfWork;
		33	//import weblogic.utils.net.SocketResetException;
31	//import javax.persistence.OptimisticLockException;	34	//import javax.persistence.OptimisticLockException;
32		35
33	/**	36	/**
Lines 49-54 Link Here
49	* Numbers greater than this are encountered in scientific, cryptographic and nanosecond time sensitive calculations.	52	* Numbers greater than this are encountered in scientific, cryptographic and nanosecond time sensitive calculations.
50	*/	53	*/
51	private static final Long MAX_BIGINTEGER_IN_SQL = Long.MAX_VALUE;	54	private static final Long MAX_BIGINTEGER_IN_SQL = Long.MAX_VALUE;
		55	/** Get number of (hyperthreaded + real) cores. IE: p630 with HT=2, Core2 E8400=2 and Core i7-920 = 8 */
		56	public static final int CORES = Runtime.getRuntime().availableProcessors() << 0;
52		57
53	// TODO: we need to move all this to a properties file	58	// TODO: we need to move all this to a properties file
54	/** this is the index of the current server - for use in the maps and lists below */	59	/** this is the index of the current server - for use in the maps and lists below */
Lines 66-72 Link Here
66	/** output cached context maps for each remote server */	71	/** output cached context maps for each remote server */
67	private Map<String, Context> rmiCachedContextMap = new HashMap<String, Context>();	72	private Map<String, Context> rmiCachedContextMap = new HashMap<String, Context>();
68	/** $Proxy remote objects */	73	/** $Proxy remote objects */
69	private Map<String, CollatzFacadeRemote> remoteObjects = new HashMap<String, CollatzFacadeRemote>();	74	private List<Map<String, CollatzFacadeRemote>> remoteObjects = new ArrayList<Map<String, CollatzFacadeRemote>>(CORES);
		75	//private Map<String, CollatzFacadeRemote> remoteObjects = new HashMap<String, CollatzFacadeRemote>();
70	/** How many processors are available (real + hyperthreaded) */	76	/** How many processors are available (real + hyperthreaded) */
71	private Map<String, Integer> availableProcessors = new HashMap<String, Integer>();	77	private Map<String, Integer> availableProcessors = new HashMap<String, Integer>();
72	/** whether the node is accepting requests or not */	78	/** whether the node is accepting requests or not */
Lines 74-79 Link Here
74	/** map of t3 protocol URLs */	80	/** map of t3 protocol URLs */
75	private Map<String, String> serverIPMap = new HashMap<String, String>();	81	private Map<String, String> serverIPMap = new HashMap<String, String>();
76		82
		83	//private List<CollatzRunnable> runnables = new ArrayList<CollatzRunnable>();
		84
77	private static final String DEFAULT_CLIENT_NAME = "default";	85	private static final String DEFAULT_CLIENT_NAME = "default";
78	// WebLogic	86	// WebLogic
79	// verify that all EE libraries available via http://download.oracle.com/docs/cd/E12840_01/wls/docs103/client/jarbuilder.html	87	// verify that all EE libraries available via http://download.oracle.com/docs/cd/E12840_01/wls/docs103/client/jarbuilder.html
Lines 97-129 Link Here
97	aTable.put(Context.INITIAL_CONTEXT_FACTORY,CONTEXT_FACTORY_NAME);	105	aTable.put(Context.INITIAL_CONTEXT_FACTORY,CONTEXT_FACTORY_NAME);
98	aTable.put(Context.PROVIDER_URL, serverT3[i]);	106	aTable.put(Context.PROVIDER_URL, serverT3[i]);
99	}	107	}
		108	for(int i=0;i<CORES;i++) {
		109	connect(i);
		110	}
100	}	111	}
101		112
102	public void connect() {	113	public void connect(int threadID) {
103	// Setup RMI Objects	114	// Setup RMI Objects
104	// Establish RMI connections to the session beans	115	// Establish RMI connections to the session beans
105	for(String aServer : serverNames) {	116	//for(String aServer : serverNames) {
		117	String aServer = serverNames.get(0);
106	Context aContext = null;	118	Context aContext = null;
107	try {	119	try {
108	// no need to set the host if on same machine	120	// no need to set the host if on same machine
109	aContext = new InitialContext(contextHashtableMap.get(aServer));	121	aContext = new InitialContext(contextHashtableMap.get(aServer));
110	rmiCachedContextMap.put(aServer, aContext);	122	rmiCachedContextMap.put(aServer, aContext);
111	System.out.println("_collatz: " + System.currentTimeMillis() + ": Context for " + aServer + " : " + aContext);	123	System.out.println("_collatz: " + System.currentTimeMillis() + ": Context for " + (aServer + threadID) + " : " + aContext);
112	// For qualified name look for weblogic log "EJB Deployed EJB with JNDI name"	124	// For qualified name look for weblogic log "EJB Deployed EJB with JNDI name"
113	Object aRemoteReference = aContext.lookup(SESSION_BEAN_REMOTE_NAME);	125	Object aRemoteReference = aContext.lookup(SESSION_BEAN_REMOTE_NAME);
114	System.out.println("_collatz: " + System.currentTimeMillis() + ": Remote Object: " + aRemoteReference);	126	System.out.println("_collatz: " + System.currentTimeMillis() + ": Remote Object: " + aRemoteReference);
115	// narrow the $proxy remote bean	127	// narrow the $proxy remote bean
116	CollatzFacadeRemote aNode = (CollatzFacadeRemote) PortableRemoteObject.narrow(aRemoteReference, CollatzFacadeRemote.class);	128	CollatzFacadeRemote aNode = (CollatzFacadeRemote) PortableRemoteObject.narrow(aRemoteReference, CollatzFacadeRemote.class);
117	remoteObjects.put(aServer, aNode);	129	Map<String, CollatzFacadeRemote> remoteObject = new HashMap<String, CollatzFacadeRemote>();
		130	remoteObject.put(aServer, aNode);
		131	remoteObjects.add(remoteObject);
118	} catch (Exception ce) {	132	} catch (Exception ce) {
119	// server down throws a javax.naming.CommunicationException inside a java.net.ConnectException	133	// server down throws a javax.naming.CommunicationException inside a java.net.ConnectException
120	ce.printStackTrace();	134	ce.printStackTrace();
121	// mark the current node as down, clear the flag in 5 min	135	// mark the current node as down, clear the flag in 5 min
122	}	136	}
123	}	137	//}
124	}	138	}
125		139
126	private CollatzFacadeRemote lookupRemoteBean(String aServer) {	140	private CollatzFacadeRemote lookupRemoteBean(String aServer, int threadID) {
127	CollatzFacadeRemote remoteBean = null;	141	CollatzFacadeRemote remoteBean = null;
128	try {	142	try {
129	Context aContext = rmiCachedContextMap.get(aServer);	143	Context aContext = rmiCachedContextMap.get(aServer);
Lines 132-233 Link Here
132	boolean remoteLookupSuccess = false;	146	boolean remoteLookupSuccess = false;
133	int lookupIterations = 0;	147	int lookupIterations = 0;
134	while(!remoteLookupSuccess && aRemoteReference == null && lookupIterations < 50) {	148	while(!remoteLookupSuccess && aRemoteReference == null && lookupIterations < 50) {
		149	System.out.println("_collatz: " + System.currentTimeMillis() + ": Thread: " + threadID + " : Context lookup for " + SESSION_BEAN_REMOTE_NAME + " from: " + aContext);
135	try {	150	try {
136	aRemoteReference = aContext.lookup(SESSION_BEAN_REMOTE_NAME);	151	aRemoteReference = aContext.lookup(SESSION_BEAN_REMOTE_NAME);
137	} catch (NameNotFoundException nnfe) {	152	} catch (NameNotFoundException nnfe) {
138	System.out.println(nnfe.getMessage());	153	System.out.println(nnfe.getMessage());
139	System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - possible redeploy in progress on central server: " + lookupIterations);	154	System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - possible redeploy in progress on central server: " + lookupIterations);
140	Thread.sleep(1000);	155	Thread.sleep(1000);
		156	} catch (CommunicationException ce) {//SocketResetException sre) {
		157	// Network was temporarily disconnected - or server went down
		158	System.out.println(ce.getMessage());
		159	System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - Network or server is temporarily down: " + lookupIterations);
		160	Thread.sleep(1000);
141	}	161	}
142	lookupIterations++;	162	lookupIterations++;
143	}	163	}
144	System.out.println("_collatz: " + System.currentTimeMillis() + ": Remote Object: " + aRemoteReference);	164	System.out.println("_collatz: " + System.currentTimeMillis() + ": Remote Object: " + aRemoteReference);
145	// narrow the $proxy remote bean	165	// narrow the $proxy remote bean
146	remoteBean = (CollatzFacadeRemote) PortableRemoteObject.narrow(aRemoteReference, CollatzFacadeRemote.class);	166	remoteBean = (CollatzFacadeRemote) PortableRemoteObject.narrow(aRemoteReference, CollatzFacadeRemote.class);
147	remoteObjects.put(aServer, remoteBean);	167	Map<String, CollatzFacadeRemote> remoteObject = new HashMap<String, CollatzFacadeRemote>();
		168	remoteObject.put(aServer, remoteBean);
		169	setRemoteObjects(threadID, remoteObject);
		170	/*} catch (ConnectException rmice) {//CommunicationException ce) {//SocketResetException sre) {
		171	// Network was temporarily disconnected - or server went down
		172	System.out.println(rmice.getMessage());
		173	System.out.println("_collatz: " + System.currentTimeMillis() + ": retry session bean lookup - Network or server is temporarily down: " + lookupIterations);
		174	Thread.sleep(1000);*/
148	} catch (Exception e) {	175	} catch (Exception e) {
149	e.printStackTrace();	176	e.printStackTrace();
150	}	177	}
151	return remoteBean;	178	return remoteBean;
152	}	179	}
153		180
154	public void processUnitOfWork() {	181	public void processUnitOfWork(int threadID) {
155	// ask for a work packet	182	// ask for a work packet
156	UnitOfWork uow = null;	183	UnitOfWork uow = null;
157	CollatzFacadeRemote collatzFacade = null;	184	CollatzFacadeRemote collatzFacade = null;
158	StringBuffer aBuffer = new StringBuffer();	185	StringBuffer aBuffer = new StringBuffer();
		186	String threadName;
159	// Endlessly generate RMI requests	187	// Endlessly generate RMI requests
160	for(;;) {	188	for(;;) {
161	try {	189	try {
162	// Send messages to entire grid in parallel if we connect to more than one server	190	// Send messages to entire grid in parallel if we connect to more than one server
163	// TODO: create Threads for each remoteObject	191	// TODO: create Threads for each remoteObject
164	for(String remoteServer : remoteObjects.keySet()) {	192	for(String remoteServer : remoteObjects.get(threadID).keySet()) {
165	collatzFacade = remoteObjects.get(remoteServer);	193	threadName = serverDNS[serverInUse] + threadID;
166	try {	194	collatzFacade = remoteObjects.get(threadID).get(remoteServer);
167	// Issue: One JVM halt will affect the entire distributed app.	195	try {
168	// don't let a node failure halt the host	196	// Issue: One JVM halt will affect the entire distributed app.
169	// this remote call can throw an EJBException wrapping a java.rmi.ConnectException	197	// don't let a node failure halt the host
170	uow = collatzFacade.requestUnitOfWork(serverDNS[serverInUse],availableProcessors.get(remoteServer));	198	// this remote call can throw an EJBException wrapping a java.rmi.ConnectException
171	if(null == uow) {	199	uow = collatzFacade.requestUnitOfWork(threadName,availableProcessors.get(remoteServer));
172	// possible redeploy	200	if(null == uow) {
173	System.out.println("_collatz: " + System.currentTimeMillis() + ": persistence not functioning on server " + serverDNS[serverInUse]);	201	// possible redeploy
174	} else {	202	System.out.println("_collatz: " + System.currentTimeMillis() + ": persistence not functioning on server " + serverDNS[serverInUse]);
175	aBuffer = new StringBuffer("_collatz: ");	203	} else {
176	aBuffer.append(System.currentTimeMillis());	204	aBuffer = new StringBuffer("_collatz: ");
177	aBuffer.append(": process UnitOfWork: ");	205	aBuffer.append(System.currentTimeMillis());
178	aBuffer.append(uow);	206	aBuffer.append(": processing UnitOfWork: ");
179	aBuffer.append(" ID#");	207	aBuffer.append(uow);
180	aBuffer.append(uow.getId());	208	aBuffer.append(" ID#");
181	aBuffer.append(" ");	209	aBuffer.append(uow.getId());
182	aBuffer.append(uow.getInitial());	210	aBuffer.append(" ");
183	aBuffer.append("-");	211	aBuffer.append(uow.getInitial());
184	aBuffer.append(uow.getExtent());	212	aBuffer.append("-");
185	aBuffer.append(" from: ");	213	aBuffer.append(uow.getExtent());
186	aBuffer.append(remoteServer);	214	aBuffer.append(" for: ");
187	System.out.println(aBuffer.toString());	215	aBuffer.append(threadName);
188	}	216	System.out.println(aBuffer.toString());
189	} catch (Exception e) {//(EJBException e) {	217	}
190	// weblogic.transaction.internal.TimedOutException: Transaction timed out after 29 seconds	218	} catch (Exception e) {//(EJBException e) {
191	// or SQLException on constraint violation	219	// weblogic.transaction.internal.TimedOutException: Transaction timed out after 29 seconds
192	// EJBException wrapping a java.rmi.ConnectException if the server is not running	220	// or SQLException on constraint violation
193	e.printStackTrace();	221	// EJBException wrapping a java.rmi.ConnectException if the server is not running
194	// mark the current node as down, clear the flag in 5 min	222	e.printStackTrace();
195	//nodeUnavailable.put(remoteServer, true);	223	// mark the current node as down, clear the flag in 5 min
196	}	224	//nodeUnavailable.put(remoteServer, true);
197	// compute collatz for the sequence	225	}
198	uow.processInterval();	226	// compute collatz for the sequence
199	// return the results to the server	227	uow.processInterval();
200	// don't cache the remote bean (it may be GC'd or redeployed)	228	Thread.yield(); //
201	collatzFacade = lookupRemoteBean(remoteServer);	229	// return the results to the server
202	boolean retry = true;	230	// don't cache the remote bean (it may be GC'd or redeployed)
203	while(retry) {	231	collatzFacade = lookupRemoteBean(remoteServer, threadID);
204	try {	232	boolean retry = true;
205	collatzFacade.postUnitOfWork(uow,retry); // possible EJBException or OptimisticLockException	233	while(retry) {
206	retry = false;	234	try {
207	} catch (Exception ole) {//OptimisticLockException ole) {	235	collatzFacade.postUnitOfWork(uow,retry); // possible EJBException or OptimisticLockException
208	//System.out.println(ole.getMessage());	236	retry = false;
209	retry = true;	237	} catch (Exception ole) {//OptimisticLockException ole) {
210	Thread.sleep(1000);	238	//System.out.println(ole.getMessage());
211	}	239	retry = true;
212	}	240	Thread.sleep(1000);
213	aBuffer = new StringBuffer("_collatz: ");	241	}
214	aBuffer.append(System.currentTimeMillis());	242	}
215	aBuffer.append(": results sent to server after ");	243	aBuffer = new StringBuffer("_collatz: ");
216	aBuffer.append(uow.getEndTimestamp() - uow.getStartTimestamp());	244	aBuffer.append(System.currentTimeMillis());
217	aBuffer.append(" ms @ ");	245	aBuffer.append(": results sent to server after ");
218	aBuffer.append(uow.getMIPS());	246	aBuffer.append(uow.getEndTimestamp() - uow.getStartTimestamp());
219	aBuffer.append(" MIPS");	247	aBuffer.append(" ms @ ");
220	System.out.println(aBuffer.toString());	248	aBuffer.append(uow.getMIPS());
221	}	249	aBuffer.append(" MIPS");
		250	System.out.println(aBuffer.toString());
		251	}
222	} catch (Exception e) {	252	} catch (Exception e) {
223	e.printStackTrace();	253	e.printStackTrace();
224	try {	254	try {
225	Thread.sleep(10000);	255	Thread.sleep(10000);
226	} catch (Exception ex) { }	256	} catch (Exception ex) { }
227	}	257	}
228	}	258	}
229	}	259	}
230		260
		261	protected void startThreads(int numberOfThreads) {
		262	threadSafetyPrivate(numberOfThreads);
		263	}
		264
		265	protected void threadSafetyPrivate(int numberOfThreads) {
		266	List<Thread> threadList = new ArrayList<Thread>();
		267	for(int i=0; i<numberOfThreads; i++) {
		268	Thread aThread = new Thread(new CollatzRunnable(i));
		269	threadList.add(aThread);
		270	// stagger the threads so they are not in lockstep
		271	try {
		272	Thread.sleep(3000);
		273	} catch (Exception e) { }
		274	aThread.start();
		275	}
		276
		277	// Wait for [threadNumber] threads to complete before ending
		278	for(Thread aThread : threadList) {
		279	try {
		280	synchronized (aThread) {
		281	aThread.join();
		282	}
		283	} catch (InterruptedException ie_Ignored) {
		284	ie_Ignored.printStackTrace();
		285	} // The InterruptedException can be ignored
		286	}
		287	}
		288
		289	// Inner class implements Runnable instead of extending Thread directly
		290	class CollatzRunnable implements Runnable {
		291	protected int id;
		292
		293	public CollatzRunnable(int anId) {
		294	id = anId;
		295	}
		296
		297	public void run() {
		298	//connect(id);
		299	// We loop an arbitrary number of iterations inside each thread
		300	processUnitOfWork(id);
		301	}
		302	}
		303
231	/**	304	/**
232	* @param args the command line arguments	305	* @param args the command line arguments
233	*/	306	*/
Lines 245-260 Link Here
245	serverT3[serverInUse] = ip;	318	serverT3[serverInUse] = ip;
246		319
247	SEClient client = new SEClient();	320	SEClient client = new SEClient();
248	client.connect();	321	// Create and start threads
249	client.processUnitOfWork();	322	client.startThreads(CORES);
		323	//client.connect();
		324	//client.processUnitOfWork();
250	}	325	}
251		326
252	public Map<String, Hashtable<String, String>> getContextMap() { return contextHashtableMap; }	327	public Map<String, Hashtable<String, String>> getContextMap() { return contextHashtableMap; }
253	public void setContextMap(Map<String, Hashtable<String, String>> contextMap) { this.contextHashtableMap = contextMap; }	328	public void setContextMap(Map<String, Hashtable<String, String>> contextMap) { this.contextHashtableMap = contextMap; }
254	public Map<String, Context> getRmiContextMap() { return rmiCachedContextMap; }	329	public Map<String, Context> getRmiContextMap() { return rmiCachedContextMap; }
255	public void setRmiContextMap(Map<String, Context> rmiContextMap) { this.rmiCachedContextMap = rmiContextMap; }	330	public void setRmiContextMap(Map<String, Context> rmiContextMap) { this.rmiCachedContextMap = rmiContextMap; }
256	public Map<String, CollatzFacadeRemote> getRemoteObjects() { return remoteObjects; }	331	public Map<String, CollatzFacadeRemote> getRemoteObjects(int threadID) { return remoteObjects.get(threadID); }
257	public void setRemoteObjects(Map<String, CollatzFacadeRemote> remoteObjects) { this.remoteObjects = remoteObjects; }	332	public void setRemoteObjects(int threadID, Map<String, CollatzFacadeRemote> remoteObjects) { this.remoteObjects.set(threadID, remoteObjects); }
258	public Map<String, Boolean> getNodeUnavailable() { return nodeUnavailable; }	333	public Map<String, Boolean> getNodeUnavailable() { return nodeUnavailable; }
259	public void setNodeUnavailable(Map<String, Boolean> nodeUnavailable) { this.nodeUnavailable = nodeUnavailable; }	334	public void setNodeUnavailable(Map<String, Boolean> nodeUnavailable) { this.nodeUnavailable = nodeUnavailable; }
260	public Map<String, String> getServerIPMap() { return serverIPMap; }	335	public Map<String, String> getServerIPMap() { return serverIPMap; }