Sensix Wireless Sensor Network Middleware    Home    Project Page    Download    CVS  Hosted by

/home/brennan/Software/sensix/source/C^4/corba/gov/lanl/isr/sensix/discovery/DiscoveryService.java Go to the documentation of this file. /* * Copyright 2008. Los Alamos National Security, LLC. This material * was produced under U.S. Government contract DE-AC52-06NA25396 for * Los Alamos National Laboratory (LANL), which is operated by Los * Alamos National Security, LLC, for the Department of Energy. The * U.S. Government has rights to use, reproduce, and distribute this * software. NEITHER THE GOVERNMENT NOR LOS ALAMOS NATIONAL SECURITY, * LLC, MAKES ANY WARRANTY, EXPRESS OR IMPLIED, OR ASSUMES ANY LEGAL * LIABILITY FOR THE USE OF THIS SOFTWARE. If software is modified to * produce derivative works, such modified software should be clearly * marked, so as not to confuse it with the version available from LANL. * * Additionally, this program is free software; you can redistribute * it and/or modify it under the terms of the GNU General Public * License as published by the Free Software Foundation; version 2 of * the License. Accordingly, this program is distributed in the hope * it will be useful, but WITHOUT ANY WARRANTY; without even the * implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR * PURPOSE. See the GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. */ package gov.lanl.isr.sensix.discovery; import java.net.*; import java.io.*; import java.util.*; import org.omg.CORBA.*; import org.omg.PortableServer.*; import com.sun.corba.se.impl.orbutil.*; import gov.lanl.isr.sensix.*; public class DiscoveryService extends SelfPOA { public static final int DISCOVERY_PORT = 2999; public static final String MCAST_ADDR_IPv6 = "FF18::178"; public static final String MCAST_ADDR_IPv4 = "224.0.0.178"; public static final String MCAST_ADDR = MCAST_ADDR_IPv6; public static final boolean INSTRUMENT = true; public static final boolean DUMP_TO_FILE = false; public static final int MEMORY_LIMIT = 8096; public static final int HEADER_SIZE = 8; // NB: the following are for wired networking, adjust as needed public static final int TX_TIMEOUT = 12; // max ms between hdr & body public static final int RETRY_ITVL = 4; // tx retry buffer time (ms) public static final int EMPTY_TIMEOUT = 1000; // ms block on recv public static final int MAX_PARENTS = 128; public static final int MAX_CHILDREN = 512; public static final int MAX_SIBLINGS = 128; public static final int MAX_PEERS = (MAX_CHILDREN + MAX_SIBLINGS); public static final int MAX_CAPABLE = 32; public static byte[] i2b(int i) { // locally always BE return i2b(i, 4); } public static byte[] i2b(int i, int size) { byte[] b = new byte[size]; for (int j = 0; j < size; j++) b[j] = (byte)(i >> (8 * ((size - 1) - j))); return b; } public static int b2i(byte b[]) { // BE conversion return b2i(b, 4); } public static int b2i(byte b[], int size) { int ret = 0; for (int j = 0; j < size; j++) ret += ((b[j] & 0x00000000ff) << (8 * ((size - 1) - j))); return ret; } public static int b2i_le(byte b[]) { // LE conversion return b2i_le(b, 4); } public static int b2i_le(byte b[], int size) { int ret = 0; for (int j = 0; j < size; j++) ret += ((b[j] & 0x00000000ff) << (8 * j)); return ret; } public static byte[] announceHeader() { byte[] hdr = new byte[HEADER_SIZE]; hdr[0] = (byte)('D' & 0xff); hdr[1] = (byte)0x10; hdr[2] = (byte)0x00; hdr[3] = (byte)0x01; byte[] len_part = DiscoveryService.i2b(5); for (int i = 0; i < 4; i++) hdr[4 + i] = len_part[i]; return hdr; } public static byte[] reportHeader() { byte[] hdr = new byte[HEADER_SIZE]; hdr[0] = (byte)('D' & 0xff); hdr[1] = (byte)0x10; hdr[2] = (byte)0x00; hdr[3] = (byte)0x02; byte[] len_part = DiscoveryService.i2b(6); for (int i = 0; i < 4; i++) hdr[4 + i] = len_part[i]; return hdr; } public static byte[] requireHeader() { byte[] hdr = new byte[HEADER_SIZE]; hdr[0] = (byte)('D' & 0xff); hdr[1] = (byte)0x10; hdr[2] = (byte)0x00; hdr[3] = (byte)0x03; byte[] len_part = DiscoveryService.i2b(10); for (int i = 0; i < 4; i++) hdr[4 + i] = len_part[i]; return hdr; } public static byte[] shareHeader(Share shr) { byte[] ior_part = shr.ior.getBytes(); byte[] hdr = new byte[HEADER_SIZE]; hdr[0] = (byte)('D' & 0xff); hdr[1] = (byte)0x10; hdr[2] = (byte)0x00; hdr[3] = (byte)0x04; byte[] len_part = DiscoveryService.i2b(6 + ior_part.length); for (int i = 0; i < 4; i++) hdr[4 + i] = len_part[i]; return hdr; } public static byte[] loadAnnounce(Announce ann) { byte[] msg = new byte[5]; byte[] id_part = DiscoveryService.i2b(ann.nodeIdent); for (int i = 0; i < 4; i++) msg[i] = id_part[i]; msg[4] = ann.hLevel; return msg; } public static byte[] loadReport(Report rpt) { byte[] msg = new byte[6]; byte[] id_part = DiscoveryService.i2b(rpt.nodeIdent); for (int i = 0; i < 4; i++) msg[i] = id_part[i]; msg[4] = rpt.hLevel; msg[5] = rpt.cType; return msg; } public static byte[] loadRequire(Require req) { byte[] msg = new byte[10]; byte[] id_part = DiscoveryService.i2b(req.nodeIdent); for (int i = 0; i < 4; i++) msg[i] = id_part[i]; msg[4] = req.hLevel; msg[5] = req.cType; byte[] target_part = DiscoveryService.i2b(req.targetIdent); for (int i = 0; i < 4; i++) msg[6 + i] = target_part[i]; return msg; } public static byte[] loadShare(Share shr) { byte[] ior_part = shr.ior.getBytes(); byte[] msg = new byte[14 + ior_part.length]; byte[] id_part = DiscoveryService.i2b(shr.nodeIdent); for (int i = 0; i < 4; i++) msg[i] = id_part[i]; msg[4] = shr.hLevel; msg[5] = shr.cType; for (int i = 0; i < ior_part.length; i++) msg[6 + i] = ior_part[i]; return msg; } private class Listener extends Thread { private DiscoveryService serv; private MulticastSocket socket; public Listener(DiscoveryService ds, MulticastSocket s) { serv = ds; socket = s; } public void run() { List<Integer> heard = new ArrayList<Integer>(); boolean keep_going; synchronized(serv) { keep_going = serv.running; } while (keep_going) { DiscoverMessage in = null; DiscoverMessageHeader hdr = null; try { // listen for announcements byte[] raw_hdr = new byte[DiscoveryService.HEADER_SIZE]; DatagramPacket packet = new DatagramPacket(raw_hdr, raw_hdr.length); socket.setSoTimeout(EMPTY_TIMEOUT); socket.receive(packet); SocketAddress sender = packet.getSocketAddress(); hdr = new DiscoverMessageHeader(raw_hdr); byte[] msg = new byte[hdr.announceSize]; packet = new DatagramPacket(msg, hdr.announceSize); socket.connect(sender); socket.setSoTimeout(TX_TIMEOUT); socket.receive(packet); socket.disconnect(); in = new DiscoverMessage(hdr, msg); serv.log(Logger.LOG.DEBUG, "Discovery msg ID:" + in.nodeIdent + " level:" + in.hLevel + " type:" + in.cType + " ior:" + in.ior); } catch (SocketTimeoutException e) { socket.disconnect(); synchronized(serv) { keep_going = serv.running; } continue; } catch (DiscoverMessageException dme) { serv.log(Logger.LOG.ERROR, dme.toString()); synchronized(serv) { keep_going = serv.running; } continue; } catch (IOException e) { serv.log(Logger.LOG.ERROR, "Discovery listener IO exception"); e.printStackTrace(System.err); synchronized(serv) { keep_going = serv.running; } continue; } if (hdr != null && in != null) { if (in.nodeIdent == node_id) { serv.log(Logger.LOG.DEBUG, "Dropped my own discover msg"); synchronized(serv) { keep_going = serv.running; } continue; } switch (hdr.announceType) { case 1: if (in.hLevel >= h_level) { if (in.hLevel == h_level) { boolean found = false; for (int i = 0; i < serv.siblings_size; i++) { if (serv.siblings_seen[i] == in.nodeIdent) { found = true; break; } } if (found == false && serv.siblings_size < MAX_SIBLINGS) { serv.siblings_seen[serv.siblings_size] = in.nodeIdent; serv.siblings_size++; serv.log(Logger.LOG.DEBUG, "Added sibling"); } } else { boolean found = false; for (int i = 0; i < serv.ancestors_size; i++) { if (serv.ancestors_seen[i] == in.nodeIdent) { found = true; break; } } if (found == false && serv.ancestors_size < MAX_PARENTS) { serv.ancestors_seen[serv.ancestors_size] = in.nodeIdent; serv.ancestors_size++; serv.log(Logger.LOG.DEBUG, "Added ancestor"); } } // ancestor or sibling: send capabilities for (int i = 0; i < serv.refs_size; i++) { byte type = serv.references[i].type; Report out = new Report(node_id, h_level, type); serv.sendDiscovery(out); } } else if (in.hLevel == (h_level - 1)) { // only immediate descendants boolean found = false; for (int i = 0; i < serv.descendants_size; i++) { if (serv.descendants_seen[i] == in.nodeIdent) { found = true; break; } } if (found == false && serv.descendants_size < MAX_CHILDREN) { serv.descendants_seen[serv.descendants_size] = in.nodeIdent; serv.descendants_size++; serv.log(Logger.LOG.DEBUG, "Added descendant"); } // descendant: let them know I'm here Announce out = new Announce(node_id, h_level); serv.sendDiscovery(out); } break; case 2: { if (in.hLevel == (h_level - 1)) { // only immediate descendants boolean found = false; for (int i = 0; i < serv.desc_db_size; i++) { if (serv.desc_database[i].id == in.nodeIdent && serv.desc_database[i].type == in.cType){ found = true; break; } } if (found == false) { int idx = serv.desc_db_size; serv.desc_database[idx] = new ObjectReference(); serv.desc_database[idx].id = in.nodeIdent; serv.desc_database[idx].type = in.cType; serv.desc_database[idx].ior = null; serv.desc_database[idx].valid = false; serv.desc_db_size++; serv.log(Logger.LOG.DEBUG, "Added descendant capability " + SenseUtil.capabilityToString(in.cType)); } } else if (in.hLevel == h_level) { boolean found = false; for (int i = 0; i < serv.sib_db_size; i++) { if (serv.sib_database[i].id == in.nodeIdent && serv.sib_database[i].type == in.cType) { found = true; break; } } if (found == false) { int idx = serv.sib_db_size; serv.sib_database[idx] = new ObjectReference(); serv.sib_database[idx].id = in.nodeIdent; serv.sib_database[idx].type = in.cType; serv.sib_database[idx].ior = null; serv.sib_database[idx].valid = false; serv.sib_db_size++; serv.log(Logger.LOG.DEBUG, "Added sibling capability " + SenseUtil.capabilityToString(in.cType)); } } Require out = new Require(node_id, h_level, in.cType, in.nodeIdent); serv.sendDiscovery(out); } break; case 3: if (in.targetIdent == node_id) { for (int i = 0; i < serv.refs_size; i++) { if (serv.references[i].type == in.cType) { Share out = new Share(node_id, h_level, in.cType, serv.references[i].ior); serv.sendDiscovery(out); } } } break; case 4: if (in.hLevel == (h_level - 1)) { // only immediate descendants boolean found = false; for (int i = 0; i < serv.desc_db_size; i++) { if (serv.desc_database[i].id == in.nodeIdent && serv.desc_database[i].type == in.cType) { found = true; if (serv.desc_database[i].ior == null) { serv.desc_database[i].ior = in.ior; serv.desc_database[i].valid = true; } break; } } if (found == false) { int idx = serv.desc_db_size; serv.desc_database[idx] = new ObjectReference(); serv.desc_database[idx].id = in.nodeIdent; serv.desc_database[idx].type = in.cType; serv.desc_database[idx].ior = in.ior; serv.desc_database[idx].valid = true; serv.desc_db_size++; serv.log(Logger.LOG.DEBUG, "Added descendant IOR for " + SenseUtil.capabilityToString(in.cType)); } } else if (in.hLevel == h_level) { boolean found = false; for (int i = 0; i < serv.sib_db_size; i++) { if (serv.sib_database[i].id == in.nodeIdent && serv.sib_database[i].type == in.cType) { found = true; if (serv.sib_database[i].ior == null) { serv.sib_database[i].ior = in.ior; serv.sib_database[i].valid = true; } break; } } if (found == false) { int idx = serv.sib_db_size; serv.sib_database[idx] = new ObjectReference(); serv.sib_database[idx].id = in.nodeIdent; serv.sib_database[idx].type = in.cType; serv.sib_database[idx].ior = in.ior; serv.sib_database[idx].valid = true; serv.sib_db_size++; serv.log(Logger.LOG.DEBUG, "Added sibling IOR for " + SenseUtil.capabilityToString(in.cType)); } } break; default: break; } } synchronized(serv) { keep_going = serv.running; } } socket.close(); serv.log(Logger.LOG.DEBUG, "Shutting down Discovery"); } } public class ObjectReference { public static final int SIZE = 1024; public int id; public byte type; public String ior; public boolean valid; } /************************************************************/ /************************************************************/ public ORB orb; public POA poa; public boolean running, initialized; public ObjectReference[] references; public int refs_size; public ObjectReference[] sib_database; public int sib_db_size; public ObjectReference[] desc_database; public int desc_db_size; public int[] ancestors_seen; public int ancestors_size; public int[] siblings_seen; public int siblings_size; public int[] descendants_seen; public int descendants_size; private int node_id; private byte h_level; private OthersImpl others; private Thread launch; private Logger logger; private MulticastSocket[] mcast_sockets; private Listener[] m_threads; private int service_port; public DiscoveryService(ORB orb, int id, byte level) throws org.omg.CORBA.SystemException, org.omg.CORBA.UserException, DiscoveryException { this(orb, id, level, "DiscoveryService", (byte)0, DISCOVERY_PORT); } public DiscoveryService(ORB orb, int id, byte level, String app) throws org.omg.CORBA.SystemException, org.omg.CORBA.UserException, DiscoveryException { this(orb, id, level, app, (byte)0, DISCOVERY_PORT); } public DiscoveryService(ORB orb, int id, byte level, String app, byte debug_level) throws org.omg.CORBA.SystemException, org.omg.CORBA.UserException, DiscoveryException { this(orb, id, level, app, debug_level, DISCOVERY_PORT); } public DiscoveryService(ORB orb_val, int id, byte level, String app, byte debug_level, int port) throws org.omg.CORBA.SystemException, org.omg.CORBA.UserException, DiscoveryException { running = false; initialized = false; node_id = id; h_level = level; service_port = port; orb = orb_val; logger = new Logger(app, debug_level); others = new OthersImpl(this); references = new ObjectReference[MAX_CAPABLE]; refs_size = 0; sib_database = new ObjectReference[MAX_SIBLINGS]; sib_db_size = 0; desc_database = new ObjectReference[MAX_CHILDREN]; desc_db_size = 0; ancestors_seen = new int[MAX_PARENTS]; ancestors_size = 0; siblings_seen = new int[MAX_SIBLINGS]; siblings_size = 0; descendants_seen = new int[MAX_CHILDREN]; descendants_size = 0; POA rootPOA = POAHelper.narrow(orb.resolve_initial_references("RootPOA")); rootPOA.the_POAManager().activate(); Policy[] poaPolicy = new Policy[3]; poaPolicy[0] = rootPOA.create_lifespan_policy( LifespanPolicyValue.TRANSIENT); poaPolicy[1] = rootPOA.create_id_assignment_policy( IdAssignmentPolicyValue.SYSTEM_ID); poaPolicy[2] = rootPOA.create_servant_retention_policy( ServantRetentionPolicyValue.RETAIN); poa = rootPOA.create_POA("DiscoveryService", rootPOA.the_POAManager(), poaPolicy); poa.the_POAManager().activate(); running = true; NetworkInterface[] nifs = getAllInterfaces(); if (nifs.length < 1) throw new DiscoveryException(DiscoveryError.NoNetwork, "No external network interfaces found."); mcast_sockets = new MulticastSocket[nifs.length]; m_threads = new Listener[nifs.length]; log(Logger.LOG.INFO, "Discovery interfaces:"); for (int i = 0; i < nifs.length; i++) { log(Logger.LOG.INFO, " " + nifs[i].getName()); try { mcast_sockets[i] = new MulticastSocket(service_port); mcast_sockets[i].setReuseAddress(true); mcast_sockets[i].setNetworkInterface(nifs[i]); mcast_sockets[i].joinGroup(InetAddress.getByName(MCAST_ADDR)); m_threads[i] = new Listener(this, mcast_sockets[i]); m_threads[i].start(); } catch (SocketException e) { throw new DiscoveryException(DiscoveryError.NoNetwork, "Socket error " + e.getMessage() + " on " + MCAST_ADDR); } catch (UnknownHostException e) { throw new DiscoveryException(DiscoveryError.NoNetwork, "Unknown host " + MCAST_ADDR + " " + e.getMessage()); } catch (IOException e) { throw new DiscoveryException(DiscoveryError.NoNetwork, "IO error " + e.getMessage()); } } sendDiscovery(new Announce(node_id, h_level)); } public void log(Logger.LOG type, String msg) { logger.log(type, msg); } public void shutdown() { synchronized(this) { running = false; } try { for (int i = 0; i < m_threads.length; i++) m_threads[i].join(); } catch (InterruptedException e) {} } public ObjectReference newObjRef() { return new ObjectReference(); } private NetworkInterface[] getAllInterfaces() { try { int size = 0, idx = 0; NetworkInterface[] nifs; Enumeration<NetworkInterface> ife; ife = NetworkInterface.getNetworkInterfaces(); for (NetworkInterface nif : Collections.list(ife)) { if (nif.getName().toLowerCase().startsWith("lo") == false) size++; } if (size < 1) return null; nifs = new NetworkInterface[size]; ife = NetworkInterface.getNetworkInterfaces(); for (NetworkInterface nif : Collections.list(ife)) { if (nif.getName().toLowerCase().startsWith("lo") == false) { nifs[idx] = nif; idx++; } } return nifs; } catch (SocketException e) { log(Logger.LOG.ERROR, "Socket exception on getNetworkInterfaces"); return null; } } public void sendDiscovery(Announce ann) { byte[] hdr = announceHeader(); byte[] msg = loadAnnounce(ann); log(Logger.LOG.DEBUG, "Sending announce- ID:" + ann.nodeIdent + " level:" + ann.hLevel); sendAnnouncement(hdr, msg); } public void sendDiscovery(Report report) { byte[] hdr = reportHeader(); byte[] msg = loadReport(report); log(Logger.LOG.DEBUG, "Sending report- ID:" + report.nodeIdent + " level:" + report.hLevel + " type:" + report.cType); sendAnnouncement(hdr, msg); } public void sendDiscovery(Require require) { byte[] hdr = requireHeader(); byte[] msg = loadRequire(require); log(Logger.LOG.DEBUG, "Sending require- ID:" + require.nodeIdent + " level:" + require.hLevel + " type:" + require.cType + " target:" + require.targetIdent); sendAnnouncement(hdr, msg); } public void sendDiscovery(Share share) { byte[] hdr = shareHeader(share); byte[] msg = loadShare(share); log(Logger.LOG.DEBUG, "Sending share- ID:" + share.nodeIdent + " level:" + share.hLevel + " type:" + share.cType + " ior:" + share.ior); sendAnnouncement(hdr, msg); } private void sendAnnouncement(byte[] hdr, byte[] msg) { while (true) { long total_len = 0; DatagramPacket hpacket = null; DatagramPacket mpacket = null; try { hpacket = new DatagramPacket(hdr, hdr.length, InetAddress.getByName(MCAST_ADDR), service_port); mpacket = new DatagramPacket(msg, msg.length, InetAddress.getByName(MCAST_ADDR), service_port); total_len = (hdr.length + msg.length) * mcast_sockets.length; for (int i = 0; i < mcast_sockets.length; i++) { mcast_sockets[i].send(hpacket); mcast_sockets[i].send(mpacket); } if (INSTRUMENT) { if (DUMP_TO_FILE) { String id = java.lang.management.ManagementFactory.getRuntimeMXBean().getName(); FileWriter os = new FileWriter("jcorba." + id + ".msgct", true); os.write(total_len + " bytes\n"); os.close(); } else System.err.println(total_len + " bytes"); } break; } catch (IllegalArgumentException iae) { try { Thread.sleep(TX_TIMEOUT + RETRY_ITVL); } catch (InterruptedException ie) {} } catch (IOException e) { log(Logger.LOG.ERROR, "Discovery IO exception"); e.printStackTrace(System.err); break; } } } /************************************************************/ /************************************************************/ public byte level() { return h_level; } public int identifier() { return node_id; } public byte[] capabilities() { byte[] caps = new byte[refs_size]; for (int i = 0; i < refs_size; i++) caps[i] = references[i].type; return caps; } public void registerObject(byte capability, gov.lanl.isr.sensix.Request obj) throws DiscoveryException { registerObject(capability, obj, false); } public void registerObject(byte capability, gov.lanl.isr.sensix.Request obj, boolean override) throws DiscoveryException { boolean found = false; if (orb == null) throw new DiscoveryException(DiscoveryError.BadORB, "Discovery service ORB was lost"); if (obj == null) throw new DiscoveryException(DiscoveryError.InvalidObject, "Attempting to register a null object"); if (refs_size > MAX_CAPABLE) throw new DiscoveryException(DiscoveryError.RegistryFull, "Registry size exceeds " + MAX_CAPABLE); for (int i = 0; i < refs_size; i++) { if (references[i].type == capability) { found = true; if (orb == null) throw new DiscoveryException(DiscoveryError.BadORB, "Discovery service ORB was lost"); if (override == true) references[i].ior = orb.object_to_string(obj); else throw new DiscoveryException( DiscoveryError.AlreadyRegistered, "Prior registration found for " + SenseUtil.capabilityToString(capability) + " capability"); } } if (found == false) { references[refs_size] = new ObjectReference(); references[refs_size].id = node_id; references[refs_size].ior = orb.object_to_string(obj); references[refs_size].type = capability; references[refs_size].valid = true; refs_size++; sendDiscovery(new Report(node_id, h_level, capability)); } } /************************************************************/ public boolean findNodeInFamily(int id) throws DiscoveryException { return others.findNode(id); } public gov.lanl.isr.sensix.Request[] queryFamilyNetwork(byte capability) throws DiscoveryException { return others.queryNetwork(capability); } public gov.lanl.isr.sensix.Request queryFamilyNode(int id, byte capability) throws DiscoveryException { return others.queryNode(id, capability); } public boolean findNodeInDescendants(int id) throws DiscoveryException { return others.descendants().findNode(id); } public gov.lanl.isr.sensix.Request[] queryDescendantNetwork(byte capability) throws DiscoveryException { return others.descendants().queryNetwork(capability); } public gov.lanl.isr.sensix.Request queryDescendantNode(int id, byte capability) throws DiscoveryException { return others.descendants().queryNode(id, capability); } public byte[] descendantCapabilities() throws DiscoveryException { return others.descendants().capabilities(); } public int[] descendantNodes() throws DiscoveryException { return others.descendants().nodes(); } public boolean findNodeInSiblings(int id) throws DiscoveryException { return others.siblings().findNode(id); } public gov.lanl.isr.sensix.Request[] querySiblingNetwork(byte capability) throws DiscoveryException { return others.siblings().queryNetwork(capability); } public gov.lanl.isr.sensix.Request querySiblingNode(int id, byte capability) throws DiscoveryException { return others.siblings().queryNode(id, capability); } public byte[] siblingCapabilities() throws DiscoveryException { return others.siblings().capabilities(); } public int[] siblingNodes() throws DiscoveryException { return others.siblings().nodes(); } public boolean findNodeInAncestors(int id) throws DiscoveryException { return others.ancestors().findNode(id); } public int[] ancestorNodes() throws DiscoveryException { return others.ancestors().nodes(); } }

© 2008, Los Alamos National Security, LLC.