parallel/services/HostnameDiscoveryService.java

package appl.parallel.services;

import java.net.MalformedURLException;
import java.rmi.ConnectException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.UnknownHostException;
import java.util.Iterator;
import java.util.Vector;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import appl.ext.XuluConfig;
import appl.parallel.ComputingResource;
import appl.parallel.ComputingResourceContainer;
import appl.parallel.starter.Starter;
import appl.parallel.starter.client.StarterContainer;
import appl.parallel.starter.server.XuluServerStarter;
import appl.parallel.test.PingTestObject;

/**
 * Very simple Discovery Service that simply looks up hosts from Property <br>
 * <br>
 * <code>DiscoveryServices.hostname.hosts</code> <br>
 * <br>
 * of the {@link XuluConfig} <br>
 * <br>
 * Refreshes the hostlist on every {@link #getResources()}, if <br>
 * <br>
 * <code> DiscoveryServices.hostname.refresh</code> <br>
 * <br>
 * is set to 1, it gets the ips from Property <br>
 * <br>
 * DiscoveryService.hostname.hosts <br>
 * <br>
 * in {@link XuluConfig}
 *
 * @author Dominik Appl
 */
public class HostnameDiscoveryService implements DiscoveryService {
        private final Logger LOG = LogManager.getLogger(this.getClass().getName());

        Vector<ComputingResource> res = new Vector<ComputingResource>();

        ExecutorService executor = Executors.newCachedThreadPool();

        private String[] lookupStrings;

        private int timeout;

        private boolean firstRefresh = true;

        /** @return true
         * @see appl.parallel.services.Service#isRunning()
         */
        public boolean isRunning() {
                return true;
        }

        /**
         * @see appl.parallel.services.Service#startService()
         */
        public void startService() {
                executor = Executors.newCachedThreadPool();
        }

        /**
         * stops the service
         */
        public void stopService() {
                executor.shutdownNow();
        }

        /**
         * tries to (re-)discover every resource in Property <code>HostnameDiscoveryService.hosts</code>
         * of {@link XuluConfig}. After the time in ms specified in DiscoveryServices.timeout
         *
         */
        @SuppressWarnings("unchecked")
        public synchronized void refresh() {
                //get timeout
                initVariables();
                res.addAll(getResourcesForLookupStrings());
        }

        private void initVariables() {
                timeout = XuluConfig.getXuluConfig().getIntProperty(
                                "DiscoveryServices.timeout");
                if (timeout == 0) {
                        LOG
                                        .warn("DiscoveryServices.timeout was not set or was 0! Setting connect timeout to 500...");
                        timeout = 500;
                }
                //get IPs
                res.clear();
                String[] ips = XuluConfig.getXuluConfig().getMultiProperty(
                                "DiscoveryServices.hostname.hosts");
                if (ips.length == 0 || (ips.length == 1 && ips[0].equals(""))) {
                        LOG.warn("No Resources found in DiscoveryServices.hostname.hosts");
                        ips = new String[0];
                }
                //generate lookupStrings:
                lookupStrings = ips;
        }

        /** gets resources, but pings them first to see if they are alive. Not
         *      reachable resouces are removed.
         *
         * @see appl.parallel.services.DiscoveryService#getResources()
         */
        public Vector<ComputingResourceContainer> getResources() {
                if (firstRefresh
                                || XuluConfig.getXuluConfig().getBooleanProperty(
                                                "DiscoveryServices.hostname.refresh") == true)
                        refresh();
                firstRefresh = false;
                return appl.parallel.util.Helper.getResourceContainersForVector(res);
        }

        /**
         * @return starterContainers for {@link XuluServerStarter}s
         */
        public Vector<StarterContainer> getStarterContainers() {
                initVariables();
                String[] lookupNames = new String[lookupStrings.length];
                for (int i = 0; i < lookupStrings.length; i++)
                        lookupNames[i] = "rmi://" + lookupStrings[i] + "/XuluServerStarter";

                Vector<Starter> discoveredStarters = new Vector<Starter>();
                Vector<StarterContainer> containers = new Vector<StarterContainer>();
                // make a new Thread for each resouce
                Future[] futures = new Future[lookupStrings.length];
                for (int i = 0; i < lookupStrings.length; i++) {
                        futures[i] = executor.submit(new LookupCallable(lookupNames[i]));
                }

                // the time at which the result must be there:
                long timeoutTime = System.currentTimeMillis() + timeout;
                for (int i = 0; i < futures.length; i++) {
                        // servers previous in iterations may have caused a delay
                        // so the remaining time is calculated
                        long remainingTime = Math.max(10, timeoutTime
                                        - System.currentTimeMillis());

                        try {
                                Starter starter = (Starter) futures[i].get(remainingTime,
                                                TimeUnit.MILLISECONDS);
                                if (starter != null) {
                                        discoveredStarters.add(starter);
                                        containers.add(new StarterContainer(starter,
                                                        lookupStrings[i]));
                                        LOG.debug("Discovered Starter: " + lookupNames[i]);
                                }
                        } catch (InterruptedException e) {
                                LOG.error(e);
                                e.printStackTrace();
                        } catch (ExecutionException e) {
                                LOG.error("Error while connecting to " + lookupNames[i]
                                                + " (Connect exception)");
                        } catch (TimeoutException e) {
                                LOG.warn("Could not find Starter " + lookupNames[i]
                                                + ". Timed out after " + timeout + "ms");
                                futures[i].cancel(true);
                        }
                }
                return containers;
        }

        private Vector<ComputingResource> getResourcesForLookupStrings() {
                initVariables();
                String[] lookupNames = new String[lookupStrings.length];
                for (int i = 0; i < lookupStrings.length; i++)
                        lookupNames[i] = "rmi://" + lookupStrings[i] + "/XuluServer";

                Vector<ComputingResource> discoveredResources = new Vector<ComputingResource>(
                                lookupNames.length);
                // make a new Thread for each resource
                Future[] futures = new Future[lookupNames.length];
                for (int i = 0; i < lookupNames.length; i++) {
                        futures[i] = executor.submit(new LookupCallable(lookupNames[i]));
                }

                // the time at which the result must be there:
                long timeoutTime = System.currentTimeMillis() + timeout;
                for (int i = 0; i < futures.length; i++) {
                        // servers previous in iterations may have caused a delay
                        // so the remaining time is calculated
                        long remainingTime = Math.max(0, timeoutTime
                                        - System.currentTimeMillis());

                        try {
                                ComputingResource server = (ComputingResource) futures[i].get(
                                                remainingTime, TimeUnit.MILLISECONDS);
                                discoveredResources.add(server);
                                LOG.debug("Discovered resource: " + lookupNames[i]);
                        } catch (InterruptedException e) {
                                LOG.error(e);
                                e.printStackTrace();
                        } catch (ExecutionException e) {
                                LOG.error("Error while connecting to " + lookupNames[i], e);
                        } catch (TimeoutException e) {
                                LOG.warn("Could not add server " + lookupNames[i]
                                                + ". Timed out after " + timeout + "ms");
                                futures[i].cancel(true);
                        }
                }
                return discoveredResources;
        }

        /**
         * This class is used for lookup of remote resources. It handles all errors
         * and submits them to the LOG
         *
         * @author Dominik Appl
         */
        private class LookupCallable implements Callable {

                private final String lookupString;

                private ComputingResource server = null;

                public LookupCallable(String lookupString) {
                        this.lookupString = lookupString;

                }

                /* (non-Javadoc)
                 * @see java.util.concurrent.Callable#call()
                 */
                public Object call() throws Exception {
                        try {
                                return Naming.lookup(lookupString);
                        } catch (UnknownHostException e) {
                                LOG.warn("(Unknown Host) Could not find host " + lookupString);
                        } catch (java.rmi.NotBoundException e) {
                                LOG.warn("(Not Bound) Could not find host " + lookupString);
                        } catch (java.rmi.ConnectException e) {
                                LOG.warn("(No Connection) Could not connect to host "
                                                + lookupString);
                        }
                        return null;
                }
        }

}
1	mojays	2	package appl.parallel.services;
2
3			import java.net.MalformedURLException;
4			import java.rmi.ConnectException;
5			import java.rmi.Naming;
6			import java.rmi.NotBoundException;
7			import java.rmi.RemoteException;
8			import java.rmi.UnknownHostException;
9			import java.util.Iterator;
10			import java.util.Vector;
11			import java.util.concurrent.Callable;
12			import java.util.concurrent.ExecutionException;
13			import java.util.concurrent.ExecutorService;
14			import java.util.concurrent.Executors;
15			import java.util.concurrent.Future;
16			import java.util.concurrent.TimeUnit;
17			import java.util.concurrent.TimeoutException;
18
19			import org.apache.log4j.LogManager;
20			import org.apache.log4j.Logger;
21
22			import appl.ext.XuluConfig;
23			import appl.parallel.ComputingResource;
24			import appl.parallel.ComputingResourceContainer;
25			import appl.parallel.starter.Starter;
26			import appl.parallel.starter.client.StarterContainer;
27			import appl.parallel.starter.server.XuluServerStarter;
28			import appl.parallel.test.PingTestObject;
29
30			/**
31			* Very simple Discovery Service that simply looks up hosts from Property <br>
32			* <br>
33			* <code>DiscoveryServices.hostname.hosts</code> <br>
34			* <br>
35			* of the {@link XuluConfig} <br>
36			* <br>
37			* Refreshes the hostlist on every {@link #getResources()}, if <br>
38			* <br>
39			* <code> DiscoveryServices.hostname.refresh</code> <br>
40			* <br>
41			* is set to 1, it gets the ips from Property <br>
42			* <br>
43			* DiscoveryService.hostname.hosts <br>
44			* <br>
45			* in {@link XuluConfig}
46			*
47			* @author Dominik Appl
48			*/
49			public class HostnameDiscoveryService implements DiscoveryService {
50			private final Logger LOG = LogManager.getLogger(this.getClass().getName());
51
52			Vector<ComputingResource> res = new Vector<ComputingResource>();
53
54			ExecutorService executor = Executors.newCachedThreadPool();
55
56			private String[] lookupStrings;
57
58			private int timeout;
59
60			private boolean firstRefresh = true;
61
62			/** @return true
63			* @see appl.parallel.services.Service#isRunning()
64			*/
65			public boolean isRunning() {
66			return true;
67			}
68
69			/**
70			* @see appl.parallel.services.Service#startService()
71			*/
72			public void startService() {
73			executor = Executors.newCachedThreadPool();
74			}
75
76			/**
77			* stops the service
78			*/
79			public void stopService() {
80			executor.shutdownNow();
81			}
82
83			/**
84			* tries to (re-)discover every resource in Property <code>HostnameDiscoveryService.hosts</code>
85			* of {@link XuluConfig}. After the time in ms specified in DiscoveryServices.timeout
86			*
87			*/
88			@SuppressWarnings("unchecked")
89			public synchronized void refresh() {
90			//get timeout
91			initVariables();
92			res.addAll(getResourcesForLookupStrings());
93			}
94
95			private void initVariables() {
96			timeout = XuluConfig.getXuluConfig().getIntProperty(
97			"DiscoveryServices.timeout");
98			if (timeout == 0) {
99			LOG
100			.warn("DiscoveryServices.timeout was not set or was 0! Setting connect timeout to 500...");
101			timeout = 500;
102			}
103			//get IPs
104			res.clear();
105			String[] ips = XuluConfig.getXuluConfig().getMultiProperty(
106			"DiscoveryServices.hostname.hosts");
107			if (ips.length == 0 \|\| (ips.length == 1 && ips[0].equals(""))) {
108			LOG.warn("No Resources found in DiscoveryServices.hostname.hosts");
109			ips = new String[0];
110			}
111			//generate lookupStrings:
112			lookupStrings = ips;
113			}
114
115			/** gets resources, but pings them first to see if they are alive. Not
116			* reachable resouces are removed.
117			*
118			* @see appl.parallel.services.DiscoveryService#getResources()
119			*/
120			public Vector<ComputingResourceContainer> getResources() {
121			if (firstRefresh
122			\|\| XuluConfig.getXuluConfig().getBooleanProperty(
123			"DiscoveryServices.hostname.refresh") == true)
124			refresh();
125			firstRefresh = false;
126			return appl.parallel.util.Helper.getResourceContainersForVector(res);
127			}
128
129			/**
130			* @return starterContainers for {@link XuluServerStarter}s
131			*/
132			public Vector<StarterContainer> getStarterContainers() {
133			initVariables();
134			String[] lookupNames = new String[lookupStrings.length];
135			for (int i = 0; i < lookupStrings.length; i++)
136			lookupNames[i] = "rmi://" + lookupStrings[i] + "/XuluServerStarter";
137
138			Vector<Starter> discoveredStarters = new Vector<Starter>();
139			Vector<StarterContainer> containers = new Vector<StarterContainer>();
140			// make a new Thread for each resouce
141			Future[] futures = new Future[lookupStrings.length];
142			for (int i = 0; i < lookupStrings.length; i++) {
143			futures[i] = executor.submit(new LookupCallable(lookupNames[i]));
144			}
145
146			// the time at which the result must be there:
147			long timeoutTime = System.currentTimeMillis() + timeout;
148			for (int i = 0; i < futures.length; i++) {
149			// servers previous in iterations may have caused a delay
150			// so the remaining time is calculated
151			long remainingTime = Math.max(10, timeoutTime
152			- System.currentTimeMillis());
153
154			try {
155			Starter starter = (Starter) futures[i].get(remainingTime,
156			TimeUnit.MILLISECONDS);
157			if (starter != null) {
158			discoveredStarters.add(starter);
159			containers.add(new StarterContainer(starter,
160			lookupStrings[i]));
161			LOG.debug("Discovered Starter: " + lookupNames[i]);
162			}
163			} catch (InterruptedException e) {
164			LOG.error(e);
165			e.printStackTrace();
166			} catch (ExecutionException e) {
167			LOG.error("Error while connecting to " + lookupNames[i]
168			+ " (Connect exception)");
169			} catch (TimeoutException e) {
170			LOG.warn("Could not find Starter " + lookupNames[i]
171			+ ". Timed out after " + timeout + "ms");
172			futures[i].cancel(true);
173			}
174			}
175			return containers;
176			}
177
178			private Vector<ComputingResource> getResourcesForLookupStrings() {
179			initVariables();
180			String[] lookupNames = new String[lookupStrings.length];
181			for (int i = 0; i < lookupStrings.length; i++)
182			lookupNames[i] = "rmi://" + lookupStrings[i] + "/XuluServer";
183
184			Vector<ComputingResource> discoveredResources = new Vector<ComputingResource>(
185			lookupNames.length);
186			// make a new Thread for each resource
187			Future[] futures = new Future[lookupNames.length];
188			for (int i = 0; i < lookupNames.length; i++) {
189			futures[i] = executor.submit(new LookupCallable(lookupNames[i]));
190			}
191
192			// the time at which the result must be there:
193			long timeoutTime = System.currentTimeMillis() + timeout;
194			for (int i = 0; i < futures.length; i++) {
195			// servers previous in iterations may have caused a delay
196			// so the remaining time is calculated
197			long remainingTime = Math.max(0, timeoutTime
198			- System.currentTimeMillis());
199
200			try {
201			ComputingResource server = (ComputingResource) futures[i].get(
202			remainingTime, TimeUnit.MILLISECONDS);
203			discoveredResources.add(server);
204			LOG.debug("Discovered resource: " + lookupNames[i]);
205			} catch (InterruptedException e) {
206			LOG.error(e);
207			e.printStackTrace();
208			} catch (ExecutionException e) {
209			LOG.error("Error while connecting to " + lookupNames[i], e);
210			} catch (TimeoutException e) {
211			LOG.warn("Could not add server " + lookupNames[i]
212			+ ". Timed out after " + timeout + "ms");
213			futures[i].cancel(true);
214			}
215			}
216			return discoveredResources;
217			}
218
219			/**
220			* This class is used for lookup of remote resources. It handles all errors
221			* and submits them to the LOG
222			*
223			* @author Dominik Appl
224			*/
225			private class LookupCallable implements Callable {
226
227			private final String lookupString;
228
229			private ComputingResource server = null;
230
231			public LookupCallable(String lookupString) {
232			this.lookupString = lookupString;
233
234			}
235
236			/* (non-Javadoc)
237			* @see java.util.concurrent.Callable#call()
238			*/
239			public Object call() throws Exception {
240			try {
241			return Naming.lookup(lookupString);
242			} catch (UnknownHostException e) {
243			LOG.warn("(Unknown Host) Could not find host " + lookupString);
244			} catch (java.rmi.NotBoundException e) {
245			LOG.warn("(Not Bound) Could not find host " + lookupString);
246			} catch (java.rmi.ConnectException e) {
247			LOG.warn("(No Connection) Could not connect to host "
248			+ lookupString);
249			}
250			return null;
251			}
252			}
253
254			}