parallel/services/HostnameDiscoveryService.java

package appl.parallel.services;

import java.rmi.Naming;
import java.rmi.UnknownHostException;
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;

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