experserver.java

MANTIS是由科罗拉多大学开发的传感器网络嵌入式操作系统。 这是mantis的0.9.5版本的源码。

            if (print)
                System.out.println();
        }
        
        /* Same as above, but for nodes instead of shells. */
        void sendToAllNodes(Collection<ShellConnection> all, byte command, boolean print) throws InterruptedException
        {
            for (Iterator<ShellConnection> i = all.iterator(); i.hasNext(); ) {
                ShellConnection sc = i.next();
                if (print)
                    System.out.print(" "+(sc.ui==null? "<null>" : Integer.toString(sc.ui.node)));
                synchronized (sc) {
                    sc.setAck(-1);
                    while (sc.getAck() != command) {
                        sc.sendNodeCommand(command);
                        sc.wait(5000);
                    }
                }
            }
            if (print)
                System.out.println();
        }
        
        /* Run an experiment many times.
         * Vary size from 'start' to 'stop' (inclusive) in 'step' increments.
         * Repeat this 'reps' times.
         * We can start anywhere in the sequence by using the startSize and startIter parameters.
         * This function may be interrupted.
         */
        void runMany(ShellConnection seed, ArrayList<ShellConnection> all, int start, int stop, int step, int reps,
                     int startSize, int startIter) throws IOException, InterruptedException
        {
            StringWriter sw = new StringWriter();
            PrintWriter fw = new PrintWriter(sw);
            StringBuffer fname = sw.getBuffer();
            
            startSize = Math.max(startSize, start);
            startIter = Math.max(startIter, 0);
            
            for (int r=startIter; r<reps; r++)
            {
                for (int s = (r==startIter? startSize : start); s<=stop; s += step)
                {
                    fname.setLength(0);
                    fw.printf("e1_1_%02d_%02d.log", s, r);
                    //fw.printf("e4_%02d_%02d.log", s, r);
                    
                    PrintWriter pw = new PrintWriter(new FileWriter(fname.toString()));
                    setLogStream(pw);
                    setLogging(true);              

                    experiment1(seed, all, true, s);
                    //experiment4(seed, all, s);

                    setLogging(false);              
                }
            }
        }
        
        /* Run a single experiment with the given node as a seed.
         * An update is started by changing the version number on the seed.
         * If bumpSize is true change the image size as well.
         * If newSize is positive use that as the new image size, otherwise increase the size by one page.
         * TODO give this method knowledge of which protocol it is testing
         */
    	void experiment1(ShellConnection seed, ArrayList<ShellConnection> all, boolean bumpSize, int newSize) throws InterruptedException
    	{
            System.out.println("Running experiment:");
            // Ping all the nodes to make sure they're all still connected
            // I've had some USB-serial convertors cause aqshell to drop the serial connection, even though the node keeps running
            System.out.println("\tTest connections: ");
            System.out.print("\t");
            sendToAllNodes(all, AQSHELL_GETVERSION, true);
            // Make sure the shells are not recording stats
            System.out.println("\tStop recording stats.");
            sendToAllShells(all, AQSHELL_STOPSTATS, false);
            System.out.println("\tClear stats.");
            // Clear any old stats
            sendToAllShells(all, AQSHELL_CLEARSTATS, false);
            // Now we can start recording stats with a blank slate
            System.out.println("\tStart recording stats.");
            sendToAllShells(all, AQSHELL_STARTSTATS, false);
            // Clear the finish time on all target nodes
            for (int i = all.size()-1; i>=0; --i) {
            	ShellConnection sc = all.get(i);
            	synchronized (sc) {
                    if (!sc.isTarget) continue;
            		sc.finishTime = 0;
            	}
            }
            long startTime = 0;
            long finishTime = 0;
            // Find the version number on the seed node
            int version = seed.getVersion();
            int size = seed.size;
            System.out.println("\tOld version is "+version);
            // Increment the version, this will cause a new update to start
            version++;
            // Are we changing the size, too?
            if (bumpSize) {
                System.out.println("\tOld size is "+size);
                if (newSize < 0)
                    // Default is one page more than current size
                    size++;
                else 
                    size = newSize;
                // Wait for the node to ACK the new version and size, resend every 5 seconds
                synchronized (seed) {
                    seed.setAck(-1);
                    while (seed.getAck() != AQSHELL_SETIMAGESIZE) {
                        seed.setImageSize(size, version);
                        seed.wait(5000);
                    }
                    startTime = seed.versionTime;
                }
                System.out.println("\tNew size is "+size);
            } else {
                // Not changing size
                // Wait for the node to ACK the new version, resend every 5 seconds
                synchronized (seed) {
                    seed.setAck(-1);
                    while (seed.getAck() != AQSHELL_SETVERSION) {
                        seed.setVersion(version);
                        seed.wait(5000);
                    }
                    startTime = seed.versionTime;
                }
            }
            System.out.println("\tNew version is "+version);
            
            // Wait until all targets have finish downloading the image
            System.out.println("\tWaiting for target nodes to finish.");
            System.out.print("\t");
            for (int i = all.size()-1; i>=0; --i) {
                ShellConnection sc = all.get(i);
            	synchronized (sc) {
                    if (!sc.isTarget) continue;
            		while (sc.finishTime == 0) {
            			sc.wait();
            		}
            		finishTime = Math.max(finishTime, sc.finishTime);
            	}
            }
            System.out.println();
            System.out.println("\tTotal programming time: "+((finishTime-startTime)/1000) + " seconds");
            System.out.println("\tWaiting for network to be quiet.");
            // Wait until no node has sent a profile, request, or data packet for the last 30 seconds
            // We do this because Deluge will keep updating forwarding nodes,
            // even if all the target nodes have completed
            for (int i = all.size()-1; i>=0; --i) {
                ShellConnection sc = all.get(i);
            	synchronized (sc) {
            		if (sc.allquiet) continue;
            		sc.wait();
            	}
            	// Start over from the beginning, since previously quiet nodes could have broken their silences
            	i = all.size()-1;
            }
            System.out.println("\tStop recording stats.");
            sendToAllShells(all, AQSHELL_STOPSTATS, false);
            // Ping all the nodes to make sure we didn't lose one during the experiment
            System.out.println("\tTest connections again.");
            System.out.print("\t");
            sendToAllNodes(all, AQSHELL_GETVERSION, true);
            // Get the stats and print them to the log
            System.out.println("\tRetrieve stats.");
            sendToAllShells(all, AQSHELL_GETSTATS, false);
            System.out.println("\tDone.");
    	}
        
        /* Run a single experiment with the given node as a seed.
         * An update is started by changing the version number on the seed.
         * If bumpSize is true change the image size as well.
         * If newSize is positive use that as the new image size, otherwise increase the size by one page.
         * TODO give this method knowledge of which protocol it is testing
         */
    	void experiment4(ShellConnection seed, ArrayList<ShellConnection> all, int newSize) throws InterruptedException
    	{
            System.out.println("Running experiment:");
            // Ping all the nodes to make sure they're all still connected
            // I've had some USB-serial convertors cause aqshell to drop the serial connection, even though the node keeps running
            System.out.println("\tTest connections: ");
            System.out.print("\t");
            sendToAllNodes(all, AQSHELL_GETVERSION, true);
            // Make sure the shells are not recording stats
            System.out.println("\tStop recording stats.");
            sendToAllShells(all, AQSHELL_STOPSTATS, false);
            System.out.println("\tClear stats.");
            // Clear any old stats
            sendToAllShells(all, AQSHELL_CLEARSTATS, false);
            // Now we can start recording stats with a blank slate
            System.out.println("\tStart recording stats.");
            sendToAllShells(all, AQSHELL_STARTSTATS, false);
            // Clear the finish time on all target nodes
            for (int i = all.size()-1; i>=0; --i) {
            	ShellConnection sc = all.get(i);
            	synchronized (sc) {
                    if (!sc.isTarget) continue;
            		sc.finishTime = 0;
            	}
            }
            long startTime = 0;
            long finishTime = 0;
            
            // Set the cache size on all forwarding nodes
            System.out.println("Setting the cache size on forwarding nodes.");
            for (int i = all.size()-1; i>=0; --i) {
            	ShellConnection sc = all.get(i);
            	synchronized (sc) {
                    if (sc.isTarget || seed == sc) continue;
                    sc.setAck(-1);
                    do {
                    	sc.setCacheSize(newSize, 1);
                    	sc.wait(5000);
                    } while (sc.getAck() != AQSHELL_SETCACHESIZE);
            	}
            }
            
            // Find the version number on the seed node
            int version = seed.getVersion();
            int size = seed.size;
            System.out.println("\tOld version is "+version);
            // Increment the version, this will cause a new update to start
            version++;
            // Wait for the node to ACK the new version, resend every 5 seconds
            synchronized (seed) {
                seed.setAck(-1);
                while (seed.getAck() != AQSHELL_SETVERSION) {
                    seed.setVersion(version);
                    seed.wait(5000);
                }
                startTime = seed.versionTime;
            }
            System.out.println("\tNew version is "+version);
            
            // Wait until all targets have finish downloading the image
            System.out.println("\tWaiting for target nodes to finish.");
            System.out.print("\t");
            for (int i = all.size()-1; i>=0; --i) {
                ShellConnection sc = all.get(i);
            	synchronized (sc) {
                    if (!sc.isTarget) continue;
            		while (sc.finishTime == 0) {
            			sc.wait();
            		}
            		finishTime = Math.max(finishTime, sc.finishTime);
            	}
            }
            System.out.println();
            System.out.println("\tTotal programming time: "+((finishTime-startTime)/1000) + " seconds");
            System.out.println("\tStop recording stats.");
            sendToAllShells(all, AQSHELL_STOPSTATS, false);
            // Ping all the nodes to make sure we didn't lose one during the experiment
            System.out.println("\tTest connections again.");
            System.out.print("\t");
            sendToAllNodes(all, AQSHELL_GETVERSION, true);
            // Get the stats and print them to the log
            System.out.println("\tRetrieve stats.");
            sendToAllShells(all, AQSHELL_GETSTATS, false);
            System.out.println("\tDone.");
    	}
	}
}

/* Data structure for communicating with aqshell and the node. */
class AqShellPacket
{
    // aqshell adds a timestamp to all packets sent to ExperServer
	int ts_sec;    // seconds
	int ts_usec;   // microseconds
	
	int command;
	int flags;     // Tells the aqshell and the node how it should handle this packet
	int id;        // The ID of the sending node, if the packet came from a node
	int seq;       // Unused
	int length;    // Count of additional data bytes, not including the header
	byte[] data = new byte[1024];  // buffer for additional data
    // TODO I don't know why I made it so big
	
    // For treating the additional data bytes like a stream
 	ByteArrayOutputStream baos;

    /* Construct a blank packet, most likely for receiving. */
    AqShellPacket()
	{
		this.command =  ExperServer.AQSHELL_NOOP;
		this.flags = 0;
		this.id = 0xFFFF;
		this.seq = 0;
		this.length = 0;
	}
	
	public void dump()
	{
		System.out.println("Packet:");
		System.out.println("\tts_sec:  "+ts_sec);
		System.out.println("\tts_usec: "+ts_usec);
		System.out.println();
		System.out.println("\tcommand: "+command);
		System.out.println("\tflags:   "+flags);
		System.out.println("\tid:      "+id);
		System.out.println("\tseq:     "+seq);
		if (baos != null) {
			data = baos.toByteArray();
			length = data.length;
		}		
		System.out.println("\tlength:  "+length);
		System.out.println();
		System.out.print  ("\tdata:    ");
		for (int i=0; i<length; i++)
			System.out.printf(" %02x", data[i]);
		System.out.println();
		System.out.println();
	}

    /* Construct a new packet initialized to the given values. */
    AqShellPacket(int command, int flags, int id, int seq, int length)
	{
		this.command = command;
		this.flags = flags;
		this.id = id;
		this.seq = seq;
		this.length = length;
	}
	
	void recv(DataInputStream in) throws IOException
	{
        // aqshell adds a timestamp to all packets sent to ExperServer
		ts_sec = in.readInt();
		ts_usec = in.readInt();
		
		command = in.readUnsignedByte();
		flags = in.readUnsignedByte();
		id = in.readUnsignedShort();
		seq = in.readUnsignedByte();
		length = in.readUnsignedByte();
		
		if (length > 0)
			in.read(data, 0, length);
	}
	
	void send(DataOutputStream out) throws IOException
	{
        // We don't send a timestamp
		out.writeByte(command);
		out.writeByte(flags);
		out.writeShort(id);
		out.writeByte(seq);
		
		if (baos != null) {
            // If the user has created an output stream for writing the additional data,
            // we need to use it instead
			data = baos.toByteArray();
			length = data.length;
			baos.reset();
		}
		out.writeByte(length);
		if (length > 0)
			out.write(data, 0, length);
		out.flush();
	}
	
	// Get the additional data as an input stream
    DataInputStream getDataInput()
	{
		return new DataInputStream(new ByteArrayInputStream(data, 0, length));
	}
	
    // Get the additional data as an output stream
	DataOutputStream getDataOutput()
	{
		if (baos == null)
			baos = new ByteArrayOutputStream(256);
		return new DataOutputStream(baos);
	}
}