tinydbnetwork.java

来自「用于传感器网络的节点操作系统 TinyOS 结构设计非常有意思」· Java 代码 · 共 518 行 · 第 1/2 页

package net.tinyos.tinydb;

import java.util.*;
//import net.tinyos.amhandler.*;
import java.awt.event.*;
import java.awt.*;
import net.tinyos.message.*;

/** TinyDBNetwork is responsible for getting and
    sending results and queries over the network

    @author madden

*/
public class TinyDBNetwork implements  Runnable, QueryListener, MessageListener /*, KeyEventPostProcessor*/ {

    static final short ROUNDS_TO_NEIGHBORHOOD_RESET = 5; //how long before we decide the neighborhood is empty
    static final short IGNORE_AGE = 5; //results older than this many epochs should be ignored

    public static final byte QUERY_MSG_ID = 101; //message ids used by tinydb
    public static final byte DATA_MSG_ID = 100;
    public static final byte UART_MSG_ID = 1;

    public static final byte MSG_SIZE = 36;

    public static final short UART_ADDR = 0x007e;
    
    //rate at which data messages are broadcast from the root -- this is now parameterized by the
    //rate of the fastest query
    static int baseBcastInterval = Integer.MAX_VALUE; 
    static final int MIN_EPOCH_DUR = 1200;  //never send faster than this...

    EpochTimer et = new EpochTimer(32); //background timer that tracks time left per epoch per query

    private MoteIF mif;

    private Hashtable qidListeners = new Hashtable();  //qid -> Vector(ResultListener)
    private Hashtable processedListeners = new Hashtable(); //qid -> Vector(ResultListener)
    private Vector listeners = new Vector(); //Vector(ResultListener), receive all results
    
    private short neighborhoodResetCount; //how long til we reset the neighborhood hashtable
    private Hashtable neighborhood = new Hashtable(); //list of motes we've heard recently

    private boolean sendingQuery = false;
   
    Vector knownQids = new Vector(); //list of known query ids
    Vector queries = new Vector(); //list of queries, where queries[i].qid = i;  some elements may be null

    
    
    /** Constructor 
	@param mif The MoteIF used to send / receive messages from the motes 
    */
    public TinyDBNetwork(MoteIF mif) {
	Thread t = new Thread(this);

	this.mif = mif;
	
	mif.registerListener(new QueryResultMsg(38), this);
	mif.registerListener(new UartMsg(38), this);

	t.start();

    }

    
    /** Add a listener to be notified when a query result for
	the specified query id arrives 
	@param rl The ResultListener to add
	@param aggResults Does the listener want processed (e.g. combined aggregate) results, 
	                  or raw results?
        @param qid The query id this listener is interested in
    */
    public void addResultListener(ResultListener rl, boolean aggResults, int qid) {
	Vector qidV;

	if (!aggResults) {
	    qidV = (Vector)qidListeners.get(new Integer(qid));
	} else {
	    qidV = (Vector)processedListeners.get(new Integer(qid));
	}
	if (qidV == null) {
	    qidV = new Vector();
	    if (!aggResults) {
		qidListeners.put(new Integer(qid), qidV);
	    } else {
		processedListeners.put(new Integer(qid), qidV);
	    }
	}
	qidV.addElement(rl);


    }

    
    /** Add a listener to be notified when any query result
	arrives 
	@param rl The listener to register
    */
    public void addResultListener(ResultListener rl) {
	listeners.addElement(rl);
    }

    /** Remove a specific result listener 
     @param rl The listener to remove
    */
    public void removeResultListener(ResultListener rl) {
	listeners.remove(rl);
	Enumeration e = qidListeners.elements();
	Vector qidV;

	while (e.hasMoreElements()) {
	    qidV = (Vector)e.nextElement();
	    qidV.remove(rl);
	}

	e = processedListeners.elements();
	while(e.hasMoreElements()) {
	    qidV = (Vector)e.nextElement();
	    qidV.remove(rl);
	}
    }


    Vector lastEpochs = new Vector(); //vector of last epochs for all queries
    Vector lastResults = new Vector(); //vector of HashTables by group id of results for all queries

    /** Process a radio message 
     Assumes results are QueryResults -- parses them, and maintains the following
    data structures:
    - most recent epoch heard for each query
    - most recent (paritially aggregated) result for each query
    
    Notifies ResultListeners for all results every time a value arries
    Notifies ResultListeners for raw results for a pariticular query id every time a result for that query arrives
    Notifies ResultListeners for processed results from the current epoch every time a new epoch begins;  the aggregate
                                  combines results that are from the same epoch and are not obviously bogus (e.g.
				  from a ridiculous epoch number);  each group is reported in a different result message.
				  
                             
    */
    public void messageReceived(int addr, Message m) {
	try {
	  if (m instanceof UartMsg) {
	      if (TinyDBMain.debug) System.out.println ("DEBUG MSG: " + ((UartMsg)m).getString_data());
	      return;
	  }
	  QueryResultMsg qrm = (QueryResultMsg)m;
	  int qid = QueryResult.queryId(qrm);

	  if (lastEpochs.size() <= (qid+1)) {
	    lastEpochs.setSize(qid+1);
	    lastResults.setSize(qid+1);
	  }
	  Hashtable curHt = (Hashtable)lastResults.elementAt(qid);
	  QueryResult curQr;
	  Integer le = (Integer)lastEpochs.elementAt(qid);
	  int lastEpoch = (le == null)?-1:le.intValue();
	  TinyDBQuery q = queries.size() > qid?(TinyDBQuery)queries.elementAt(qid):null;
	  if (q != null) {
	    QueryResult newqr = new QueryResult(q, qrm);
	    Vector listeners;
	    Enumeration e;
	    
	    q.setActive(true); //note that we've seen some results for this query recently
	    neighborhood.put(new Integer(newqr.getSender()), new Integer(0)); //keep a list of the sensors we've heard recently

	    //send this result to listeners for all results
	    e = this.listeners.elements();
	    while (e.hasMoreElements()) {
		((ResultListener)e.nextElement()).addResult(newqr);
	    }

	    //plus listeners for unprocessed results for this query id
	    listeners = (Vector)qidListeners.get(new Integer(newqr.qid()));
	    if (listeners != null) {
		e = listeners.elements();
		while (e.hasMoreElements()) {
		    ((ResultListener)e.nextElement()).addResult(newqr);
		}
	    }
		
	    if (newqr.getRecipient() != 0) {
		if (TinyDBMain.debug) System.out.print("("+newqr.getRecipient()+"->" +newqr.getSender()+")");
		return; //not for us
	    }
	    if (newqr.epochNo() > lastEpoch + 1000) {
		if (TinyDBMain.debug) System.out.println("e");
	      return; //ignore wacky results!
	    }

	    if (q.isAgg()) {
	      if (newqr.epochNo() > lastEpoch || curHt == null) { //onto a new epoch!
		if (curHt != null) {
		  Iterator it = curHt.values().iterator();
		  while (it.hasNext()) {
		    curQr = (QueryResult)it.next();
		    listeners = (Vector)processedListeners.get(new Integer(curQr.qid()));
		    if (listeners != null) {
			e = listeners.elements();
			while (e.hasMoreElements()) {
			    ((ResultListener)e.nextElement()).addResult(curQr);
			}
		    }
		  }
		}
		curHt = new Hashtable();
		curHt.put(new Integer(newqr.group()), newqr);
		lastResults.setElementAt(curHt, qid);
		lastEpochs.setElementAt(new Integer(newqr.epochNo()), qid);
		if (TinyDBMain.debug) System.out.print("+");
	     } else if (newqr.epochNo() >= (lastEpoch - IGNORE_AGE)) { //ignore really old results
		curQr = (QueryResult)curHt.get(new Integer(newqr.group()));
		if (curQr != null)
		  curQr.mergeQueryResult(newqr);
		else 
		  curHt.put(new Integer(newqr.group()), newqr);
		//lastEpochs.setElementAt(new Integer(newqr.epochNo()), qid);
		if (TinyDBMain.debug) System.out.print(newqr.getSender());
	      } else
		  if (TinyDBMain.debug) System.out.print("$");

	    } else { //not agg
	      if (newqr.epochNo() >= (lastEpoch - IGNORE_AGE)) { //skip old results
		  if (TinyDBMain.debug) System.out.println("Result for query " + QueryResult.queryId(qrm) + ":"+ newqr.toString());
		listeners = (Vector)processedListeners.get(new Integer(newqr.qid()));
		if (listeners != null) {
		    e = listeners.elements();
		    while (e.hasMoreElements()) {
			((ResultListener)e.nextElement()).addResult(newqr);
		    }
		}
		lastEpochs.setElementAt(new Integer(newqr.epochNo()), qid);
		if (TinyDBMain.debug) System.out.print("+");
	      } else
		  if (TinyDBMain.debug)System.out.print("[$"+newqr.epochNo()+"]");
	    }	    


	  }
	} catch (ArrayIndexOutOfBoundsException e) {
	    e.printStackTrace();
	    System.out.print("-");
	}
    }

    /** Background thread used to periodically send information from the root
	down into the network;  current this information includes:
	a message index (so that children can choose root as parent)
	information about the typical number of senders during an epoch (so that children can schedule comm)
	an epoch number (per query).
    */
    public void run() {
	Message m;
	short idx = 0;
	byte curQuery = 0, qid;
	short nwSizeEstimate = 16; //initial value

	while (true) {