parallel/services/MulticastDiscoveryService.java

package appl.parallel.services;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.UnknownHostException;
import java.rmi.Naming;
import java.rmi.RemoteException;
import java.util.Iterator;
import java.util.Vector;

import org.apache.log4j.BasicConfigurator;
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.server.XuluServer;
import appl.parallel.test.PingTestObject;
import appl.parallel.util.Helper;

/**
 * Responsible for discovery of RemoteResources on the network, especially but
 * not only {@link XuluServer XuluServers}. It listens for the following
 * messages: <br>
 * <br>
 * "Hello Xulu from XuluServer" - sent by RemoteResources like
 * {@link XuluServer} when starting <br>
 * "Goodbye Xulu" - sent by RemoteResources like {@link XuluServer} when
 * stopping <br>
 * <br>
 * and sends: <br>
 * "Hello Servers" - when started If multicasting does not work try to disable
 * your firewall or better configure it to allow multicast packages. Notice
 * also, that multicasting may not work in virtual machines like MS Virtual PC
 * 2004. <Warning><b>THIS CLASS IS STILL FOR DEMONSTRATION ONLY</b></Warning>
 * 
 * @author Dominik Appl
 * @see XuluServer
 */
public class MulticastDiscoveryService implements DiscoveryService {

        // the data to send
        final String sendHello = "Hello Servers";

        // data to receive
        final String helloFromXuluServers = "Hello Xulu from XuluServer";

        final String recGoodBye = "Goodbye Xulu";

        // the Thread which waits on Server hello/goodbyes
        volatile ReceiveThread receiveThread;

        // volatile RenewalThread renewalThread;

        // the thread that checks if servers are alive
        // volatile CheckThread checkThread;

        // the data is read from XuluConfig
        InetAddress group;

        int port;

        int lease;

        MulticastSocket socket;

        boolean isRunning = false;

        private final Logger LOG = LogManager.getLogger(this.getClass().getName());

        /**
         * Constructs a new service reading port and lease time from
         * {@link appl.ext.XuluConfig}
         */
        public MulticastDiscoveryService() {
                XuluConfig config = XuluConfig.getXuluConfig();
                port = Integer.valueOf(config
                                .getProperty("DiscoveryServices.multicast.port"));
                // lease = Integer.valueOf(config
                // .getProperty("MulticastDiscoveryService.lease"));
                try {
                        group = InetAddress.getByName(config
                                        .getProperty("DiscoveryServices.multicast.group"));
                } catch (UnknownHostException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
        }

        /**
         * Constructs a new service
         * 
         * @param port
         *            port where the multicasting happens
         * @param lease
         *            timeintervall (in ms) after whi
         */
        public MulticastDiscoveryService(int port, int lease) {
                this();
                this.port = port;
                this.lease = lease;
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#startService()
         */
        public void startService() {
                try {
                        LOG.info("Discoveryservice started");
                        // All Servers must listen on the port specified, in order fo
                        // the socket to receive
                        if (socket == null)
                                socket = new MulticastSocket(port);
                        // open a multicast group and join it

                        socket.joinGroup(group);

                        socket.setSoTimeout(5000);

                        // create and send Hello world
                        DatagramPacket datagram = new DatagramPacket(sendHello.getBytes(),
                                        sendHello.length(), group, port);
                        System.out.println(group);
                        LOG.info("Send hello to all listening servers with broadcast IP "
                                        + group);

                        socket.send(datagram);

                        if (receiveThread == null) {
                                receiveThread = new ReceiveThread(socket, helloFromXuluServers,
                                                recGoodBye);
                        }
                        if (!(receiveThread.isAlive()))
                                receiveThread.start();

                } catch (UnknownHostException e) {
                        e.printStackTrace();
                } catch (IOException e) {
                        e.printStackTrace();
                }

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.DiscoveryService#getResources()
         */
        public Vector<ComputingResourceContainer> getResources() {
                return receiveThread.getResources();
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#stopService()
         */
        public void stopService() {
                receiveThread.stopThread();
                // renewalThread.stopThread();
                try {
                        Thread.currentThread().sleep(100);
                } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
                socket.close();

        }

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

        /** waits for incoming calls */

        class ReceiveThread extends Thread {

                // Resources discovered so far
                Vector<ComputingResource> discoveredResources;

                private volatile Thread receiveThread;

                boolean exit = false;

                MulticastSocket socket;

                private final String recHello2;

                private final String recGoodBye2;

                private boolean discoveryInProgress = false;

                /**
                 * @return a Vector of discovered Ressources before return, all values
                 *          are pinged and only returned if alive
                 */
                public Vector<ComputingResourceContainer> getResources() {
                        return Helper.getResourceContainersForVector(discoveredResources);
                }

                /**
                 * @param socket2
                 *            Socket which is used for listening
                 * @param helloFromXuluServers
                 *            String that equals the hello message of the server
                 * @param recGoodBye
                 *            String that equals the Goordbye message of the server
                 */
                public ReceiveThread(MulticastSocket socket, String recHello,
                                String recGoodBye) {
                        recHello2 = recHello;
                        recGoodBye2 = recGoodBye;
                        this.socket = socket;
                        discoveredResources = new Vector<ComputingResource>();
                }

                /**
                 * Safe method to stop the Thread. Use this method instead of
                 * {@link Thread}
                 */
                public void stopThread() {
                        if (receiveThread != null) {
                                exit = true;
                                // the only way to stop the Thread is to send a package so that
                                // the receive operation no longer blocks
                                DatagramPacket datagram = new DatagramPacket(
                                                "Stopping receive Thread".getBytes(),
                                                "Stopping receive Thread".length(), group, port);
                                try {
                                        socket.send(datagram);
                                        // wait on stop
                                        Thread.currentThread().sleep(100);
                                } catch (IOException e) {
                                        e.printStackTrace();
                                } catch (InterruptedException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                } finally {
                                        exit = false;
                                }
                                exit = false;

                        }

                }

                public void run() {
                        receiveThread = Thread.currentThread();
                        while (!exit) {
                                try {
                                        // receive buffer

                                        byte[] buffer = new byte[256];
                                        DatagramPacket datagram = new DatagramPacket(buffer,
                                                        buffer.length);
                                        socket.receive(datagram);
                                        // get the message in the right
                                        byte[] result = new byte[datagram.getLength()];
                                        System.arraycopy(datagram.getData(), 0, result, 0, datagram
                                                        .getLength());
                                        // check if message is from XuluServer
                                        if (new String(result).equals(helloFromXuluServers)) {
                                                LOG.debug("received hello from server "
                                                                + datagram.getAddress());

                                                String newname = "rmi:/" + datagram.getAddress()
                                                                + "/XuluServer";

                                                ComputingResource server = (ComputingResource) Naming
                                                                .lookup(newname);
                                                addRessource(server, datagram.getAddress().toString());

                                        } else if (LOG.isDebugEnabled())
                                                LOG.debug("received message |" + new String(result)
                                                                + "| from " + datagram.getAddress());

                                } catch (java.net.SocketTimeoutException e) {
                                        // catch timeout exception
                                }

                                catch (IOException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                } catch (Exception e) {
                                        if (!(e instanceof InterruptedException))
                                                e.printStackTrace();
                                }
                        }
                        System.out.println("exit Thread " + this.toString());
                }

                /**
                 * adds ressource to vector but first looks if its not already added
                 * 
                 * @param server
                 */
                private void addRessource(ComputingResource server, String IP) {
                        // check for double entries
                        for (ComputingResource s : discoveredResources) {
                                if (s.equals(server)) {
                                        LOG.debug("Not adding " + server.toString()
                                                        + " to discovered Ressources again!");
                                        return;
                                }
                        }
                        LOG.info("Adding Xulu Server: " + IP + "to discovered Ressources");
                        // try {
                        // Thread.currentThread().sleep(2000);
                        // } catch (InterruptedException e) {
                        // // TODO Auto-generated catch block
                        // e.printStackTrace();
                        // }
                        discoveredResources.add(server);
                }

                /**
                 * Gets the {@link Resource
                 */
                public void pingResources() {
                        PingTestObject po = new PingTestObject(32);
                        // cannot make a for each loop because of
                        // java.util.ConcurrentModificationException
                        int i = discoveredResources.size() - 1;
                        while (i >= 0) {
                                boolean error = true;
                                ComputingResource r = discoveredResources.get(i);
                                if (!(r == null)) {

                                        try {
                                                r.ping();
                                                error = false;
                                                LOG.debug("Ping to " + r + " was successfull!");
                                        } catch (Exception e) {
                                                e.printStackTrace();
                                        }
                                        // if not successfull remove object
                                        if (error) {
                                                LOG.debug("Ping to " + r
                                                                + " was NOT successfull! Removing ressource");

                                                discoveredResources.remove(r);
                                        }
                                }
                                i--;
                        }
                }

        }

        /**
         * simple Thread that caused rediscovery of Ressources at a given time
         * intervall **
         */

        class RenewalThread extends Thread {

                private volatile Thread thisThread;

                boolean exit = false;

                private final long timeintervall;

                private volatile ReceiveThread discoveryToRenew;

                /**
                 * invokes rediscovery on the given thread at the given time intervall
                 */
                public RenewalThread(ReceiveThread discoveryToRenew, long timeintervall) {
                        this.discoveryToRenew = discoveryToRenew;
                        this.timeintervall = timeintervall;

                }

                /**
                 * Safe method to stop this Thread. Use this method instead of
                 * {@link Thread}
                 */
                public void stopThread() {
                        if (thisThread != null)
                                exit = true;
                }

                public void run() {
                        thisThread = Thread.currentThread();
                        while (!exit) {
                                discoveryToRenew.pingResources();
                                try {
                                        Thread.currentThread().sleep(timeintervall);
                                } catch (InterruptedException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                }
                        }
                        System.out.println("exit renewal Thread");
                }

        }

        /**
         * just for testing..edit as you like
         */
        public static void main(String[] args) {
                BasicConfigurator.configure();
                MulticastDiscoveryService bro = new MulticastDiscoveryService();
                bro.startService();
                try {
                        Thread.currentThread().sleep(5000);
                } catch (InterruptedException e) {
                        e.printStackTrace();
                }
                // bro.stopService();
                while (true)
                        try {
                                Vector<ComputingResourceContainer> v = bro.getResources();
                                for (ComputingResourceContainer container : v) {
                                        System.out.print("Active Object: ");
                                        System.out.println(container.getInformation().getProperty(
                                                        "name")
                                                        + " with IP "
                                                        + container.getInformation().getProperty("IP"));
                                }
                                Thread.currentThread().sleep(5000);
                        } catch (InterruptedException e) {

                        }
        }

}
1	mojays	2	package appl.parallel.services;
2
3			import java.io.IOException;
4			import java.net.DatagramPacket;
5			import java.net.InetAddress;
6			import java.net.MulticastSocket;
7			import java.net.UnknownHostException;
8			import java.rmi.Naming;
9			import java.rmi.RemoteException;
10			import java.util.Iterator;
11			import java.util.Vector;
12
13			import org.apache.log4j.BasicConfigurator;
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.server.XuluServer;
21			import appl.parallel.test.PingTestObject;
22			import appl.parallel.util.Helper;
23
24			/**
25			* Responsible for discovery of RemoteResources on the network, especially but
26			* not only {@link XuluServer XuluServers}. It listens for the following
27			* messages: <br>
28			* <br>
29			* "Hello Xulu from XuluServer" - sent by RemoteResources like
30			* {@link XuluServer} when starting <br>
31			* "Goodbye Xulu" - sent by RemoteResources like {@link XuluServer} when
32			* stopping <br>
33			* <br>
34			* and sends: <br>
35			* "Hello Servers" - when started If multicasting does not work try to disable
36			* your firewall or better configure it to allow multicast packages. Notice
37			* also, that multicasting may not work in virtual machines like MS Virtual PC
38			* 2004. <Warning><b>THIS CLASS IS STILL FOR DEMONSTRATION ONLY</b></Warning>
39			*
40			* @author Dominik Appl
41			* @see XuluServer
42			*/
43			public class MulticastDiscoveryService implements DiscoveryService {
44
45			// the data to send
46			final String sendHello = "Hello Servers";
47
48			// data to receive
49			final String helloFromXuluServers = "Hello Xulu from XuluServer";
50
51			final String recGoodBye = "Goodbye Xulu";
52
53			// the Thread which waits on Server hello/goodbyes
54			volatile ReceiveThread receiveThread;
55
56			// volatile RenewalThread renewalThread;
57
58			// the thread that checks if servers are alive
59			// volatile CheckThread checkThread;
60
61			// the data is read from XuluConfig
62			InetAddress group;
63
64			int port;
65
66			int lease;
67
68			MulticastSocket socket;
69
70			boolean isRunning = false;
71
72			private final Logger LOG = LogManager.getLogger(this.getClass().getName());
73
74			/**
75			* Constructs a new service reading port and lease time from
76			* {@link appl.ext.XuluConfig}
77			*/
78			public MulticastDiscoveryService() {
79			XuluConfig config = XuluConfig.getXuluConfig();
80			port = Integer.valueOf(config
81			.getProperty("DiscoveryServices.multicast.port"));
82			// lease = Integer.valueOf(config
83			// .getProperty("MulticastDiscoveryService.lease"));
84			try {
85			group = InetAddress.getByName(config
86			.getProperty("DiscoveryServices.multicast.group"));
87			} catch (UnknownHostException e) {
88			// TODO Auto-generated catch block
89			e.printStackTrace();
90			}
91			}
92
93			/**
94			* Constructs a new service
95			*
96			* @param port
97			* port where the multicasting happens
98			* @param lease
99			* timeintervall (in ms) after whi
100			*/
101			public MulticastDiscoveryService(int port, int lease) {
102			this();
103			this.port = port;
104			this.lease = lease;
105			}
106
107			/*
108			* (non-Javadoc)
109			*
110			* @see appl.parallel.services.Service#startService()
111			*/
112			public void startService() {
113			try {
114			LOG.info("Discoveryservice started");
115			// All Servers must listen on the port specified, in order fo
116			// the socket to receive
117			if (socket == null)
118			socket = new MulticastSocket(port);
119			// open a multicast group and join it
120
121			socket.joinGroup(group);
122
123			socket.setSoTimeout(5000);
124
125			// create and send Hello world
126			DatagramPacket datagram = new DatagramPacket(sendHello.getBytes(),
127			sendHello.length(), group, port);
128			System.out.println(group);
129			LOG.info("Send hello to all listening servers with broadcast IP "
130			+ group);
131
132			socket.send(datagram);
133
134			if (receiveThread == null) {
135			receiveThread = new ReceiveThread(socket, helloFromXuluServers,
136			recGoodBye);
137			}
138			if (!(receiveThread.isAlive()))
139			receiveThread.start();
140
141			} catch (UnknownHostException e) {
142			e.printStackTrace();
143			} catch (IOException e) {
144			e.printStackTrace();
145			}
146
147			}
148
149			/*
150			* (non-Javadoc)
151			*
152			* @see appl.parallel.services.DiscoveryService#getResources()
153			*/
154			public Vector<ComputingResourceContainer> getResources() {
155			return receiveThread.getResources();
156			}
157
158			/*
159			* (non-Javadoc)
160			*
161			* @see appl.parallel.services.Service#stopService()
162			*/
163			public void stopService() {
164			receiveThread.stopThread();
165			// renewalThread.stopThread();
166			try {
167			Thread.currentThread().sleep(100);
168			} catch (InterruptedException e) {
169			// TODO Auto-generated catch block
170			e.printStackTrace();
171			}
172			socket.close();
173
174			}
175
176			/*
177			* (non-Javadoc)
178			*
179			* @see appl.parallel.services.Service#isRunning()
180			*/
181			public boolean isRunning() {
182			return isRunning;
183			}
184
185			/** waits for incoming calls */
186
187			class ReceiveThread extends Thread {
188
189			// Resources discovered so far
190			Vector<ComputingResource> discoveredResources;
191
192			private volatile Thread receiveThread;
193
194			boolean exit = false;
195
196			MulticastSocket socket;
197
198			private final String recHello2;
199
200			private final String recGoodBye2;
201
202			private boolean discoveryInProgress = false;
203
204			/**
205	alfonx	33	* @return a Vector of discovered Ressources before return, all values
206	mojays	2	* are pinged and only returned if alive
207			*/
208			public Vector<ComputingResourceContainer> getResources() {
209			return Helper.getResourceContainersForVector(discoveredResources);
210			}
211
212			/**
213			* @param socket2
214			* Socket which is used for listening
215			* @param helloFromXuluServers
216			* String that equals the hello message of the server
217			* @param recGoodBye
218			* String that equals the Goordbye message of the server
219			*/
220			public ReceiveThread(MulticastSocket socket, String recHello,
221			String recGoodBye) {
222			recHello2 = recHello;
223			recGoodBye2 = recGoodBye;
224			this.socket = socket;
225			discoveredResources = new Vector<ComputingResource>();
226			}
227
228			/**
229			* Safe method to stop the Thread. Use this method instead of
230			* {@link Thread}
231			*/
232			public void stopThread() {
233			if (receiveThread != null) {
234			exit = true;
235			// the only way to stop the Thread is to send a package so that
236			// the receive operation no longer blocks
237			DatagramPacket datagram = new DatagramPacket(
238			"Stopping receive Thread".getBytes(),
239			"Stopping receive Thread".length(), group, port);
240			try {
241			socket.send(datagram);
242			// wait on stop
243			Thread.currentThread().sleep(100);
244			} catch (IOException e) {
245			e.printStackTrace();
246			} catch (InterruptedException e) {
247			// TODO Auto-generated catch block
248			e.printStackTrace();
249			} finally {
250			exit = false;
251			}
252			exit = false;
253
254			}
255
256			}
257
258			public void run() {
259			receiveThread = Thread.currentThread();
260			while (!exit) {
261			try {
262			// receive buffer
263
264			byte[] buffer = new byte[256];
265			DatagramPacket datagram = new DatagramPacket(buffer,
266			buffer.length);
267			socket.receive(datagram);
268			// get the message in the right
269			byte[] result = new byte[datagram.getLength()];
270			System.arraycopy(datagram.getData(), 0, result, 0, datagram
271			.getLength());
272			// check if message is from XuluServer
273			if (new String(result).equals(helloFromXuluServers)) {
274			LOG.debug("received hello from server "
275			+ datagram.getAddress());
276
277			String newname = "rmi:/" + datagram.getAddress()
278			+ "/XuluServer";
279
280			ComputingResource server = (ComputingResource) Naming
281			.lookup(newname);
282			addRessource(server, datagram.getAddress().toString());
283
284			} else if (LOG.isDebugEnabled())
285			LOG.debug("received message \|" + new String(result)
286			+ "\| from " + datagram.getAddress());
287
288			} catch (java.net.SocketTimeoutException e) {
289			// catch timeout exception
290			}
291
292			catch (IOException e) {
293			// TODO Auto-generated catch block
294			e.printStackTrace();
295			} catch (Exception e) {
296			if (!(e instanceof InterruptedException))
297			e.printStackTrace();
298			}
299			}
300			System.out.println("exit Thread " + this.toString());
301			}
302
303			/**
304			* adds ressource to vector but first looks if its not already added
305			*
306			* @param server
307			*/
308			private void addRessource(ComputingResource server, String IP) {
309			// check for double entries
310			for (ComputingResource s : discoveredResources) {
311			if (s.equals(server)) {
312			LOG.debug("Not adding " + server.toString()
313			+ " to discovered Ressources again!");
314			return;
315			}
316			}
317			LOG.info("Adding Xulu Server: " + IP + "to discovered Ressources");
318			// try {
319			// Thread.currentThread().sleep(2000);
320			// } catch (InterruptedException e) {
321			// // TODO Auto-generated catch block
322			// e.printStackTrace();
323			// }
324			discoveredResources.add(server);
325			}
326
327			/**
328			* Gets the {@link Resource
329			*/
330			public void pingResources() {
331			PingTestObject po = new PingTestObject(32);
332			// cannot make a for each loop because of
333			// java.util.ConcurrentModificationException
334			int i = discoveredResources.size() - 1;
335			while (i >= 0) {
336			boolean error = true;
337			ComputingResource r = discoveredResources.get(i);
338			if (!(r == null)) {
339
340			try {
341			r.ping();
342			error = false;
343			LOG.debug("Ping to " + r + " was successfull!");
344			} catch (Exception e) {
345			e.printStackTrace();
346			}
347			// if not successfull remove object
348			if (error) {
349			LOG.debug("Ping to " + r
350			+ " was NOT successfull! Removing ressource");
351
352			discoveredResources.remove(r);
353			}
354			}
355			i--;
356			}
357			}
358
359			}
360
361			/**
362			* simple Thread that caused rediscovery of Ressources at a given time
363			* intervall **
364			*/
365
366			class RenewalThread extends Thread {
367
368			private volatile Thread thisThread;
369
370			boolean exit = false;
371
372			private final long timeintervall;
373
374			private volatile ReceiveThread discoveryToRenew;
375
376			/**
377			* invokes rediscovery on the given thread at the given time intervall
378			*/
379			public RenewalThread(ReceiveThread discoveryToRenew, long timeintervall) {
380			this.discoveryToRenew = discoveryToRenew;
381			this.timeintervall = timeintervall;
382
383			}
384
385			/**
386			* Safe method to stop this Thread. Use this method instead of
387			* {@link Thread}
388			*/
389			public void stopThread() {
390			if (thisThread != null)
391			exit = true;
392			}
393
394			public void run() {
395			thisThread = Thread.currentThread();
396			while (!exit) {
397			discoveryToRenew.pingResources();
398			try {
399			Thread.currentThread().sleep(timeintervall);
400			} catch (InterruptedException e) {
401			// TODO Auto-generated catch block
402			e.printStackTrace();
403			}
404			}
405			System.out.println("exit renewal Thread");
406			}
407
408			}
409
410			/**
411			* just for testing..edit as you like
412			*/
413			public static void main(String[] args) {
414			BasicConfigurator.configure();
415			MulticastDiscoveryService bro = new MulticastDiscoveryService();
416			bro.startService();
417			try {
418			Thread.currentThread().sleep(5000);
419			} catch (InterruptedException e) {
420			e.printStackTrace();
421			}
422			// bro.stopService();
423			while (true)
424			try {
425			Vector<ComputingResourceContainer> v = bro.getResources();
426			for (ComputingResourceContainer container : v) {
427			System.out.print("Active Object: ");
428			System.out.println(container.getInformation().getProperty(
429			"name")
430			+ " with IP "
431			+ container.getInformation().getProperty("IP"));
432			}
433			Thread.currentThread().sleep(5000);
434			} catch (InterruptedException e) {
435
436			}
437			}
438
439			}