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