deluge_command.c

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

	command->command = DELUGE_COMMAND_POWER;
	command->to = -1;
	command->id = mos_node_id_get();
	command->type = DELUGE_PACKET_COMMAND;
	
	spkt.size = 1+sizeof(deluge_foot_command);
	printf_P(sPort);
	spkt.data[spkt.size++] = prompt_uint8("");
	spkt.data[spkt.size++] = DELUGE_PROTO_ID;
	
	com_send(IFACE_RADIO, &spkt);
}

static uint8_t checkCache(uint16_t id, uint8_t seq)
{
	uint8_t i;
	for (i=0; i<SEQ_CACHE_SIZE; i++)
		if (id == idCache[i] && seq == seqCache[i])
			return 1;
	idCache[nextSeq] = id;
	seqCache[nextSeq] = seq;
	nextSeq = (nextSeq + 1) % SEQ_CACHE_SIZE;
	return 0;
}

static void recvThread()
{
	com_mode(IFACE_RADIO, IF_LISTEN);
	while(1)
	{
		comBuf* pkt = com_recv(IFACE_RADIO);
		//printf("!");
		if (pkt->data[pkt->size-1] != DELUGE_PROTO_ID ||
			pkt->data[pkt->size-3] != DELUGE_PACKET_COMMAND)
		{
			com_free_buf(pkt);
			continue;
		}
		//printf("*");
		
		deluge_foot_command* command = (deluge_foot_command*)
			&pkt->data[pkt->size-2-sizeof(deluge_foot_command)];
		if (checkCache(command->id, command->seq)) {
			com_free_buf(pkt);
			continue;
		}
		//printf(".");
		
		switch (command->command)
		{
			case DELUGE_COMMAND_SENDSTATS_REPLY:
			{
				if (command->id != statsNode) break;
				uint8_t* data = (uint8_t*)packet_counts;
				memcpy(&data[pkt->data[32]], pkt->data, 32);
				
				mos_mutex_lock(&waitMutex);
				received[pkt->data[32]/32] = 1;
				mos_cond_broadcast(&waitCond);
				mos_mutex_unlock(&waitMutex);
				break;
			}
			case DELUGE_COMMAND_FINISHED:
				if (nodeFlags[command->id] & RECEIVER) {
					printf_P(sLogOn);	// turn on logging
				}
				printf_P(sFinish, command->id);
				printf_P(sShowTime);	// display time
				if (nodeFlags[command->id] & RECEIVER) {
					printf_P(sLogOff);	// turn off logging
				}
				
				mos_mutex_lock(&waitMutex);
				nodeFlags[command->id] |= FINISHED;
				if (nodeFlags[command->id] & WAIT_FINISH)
					mos_cond_broadcast(&waitCond);
				mos_mutex_unlock(&waitMutex);						
				break;
			case DELUGE_COMMAND_VERSION_REPLY:
				if (command->id != verNode) break;
				printf_P(sLogOn);	// turn on logging
				printf_P(sVerAck, command->id);
				printf_P(sShowTime);	// display time
				printf_P(sLogOff);	// turn off logging
				
				mos_mutex_lock(&waitMutex);
				ver_ack = 1;
				mos_cond_broadcast(&waitCond);
				mos_mutex_unlock(&waitMutex);						
				break;
			case DELUGE_COMMAND_NEIGHBORS_REPLY:
				break;
		}
		com_free_buf(pkt);
	}
}

static void give_up()
{
	mos_mutex_lock(&waitMutex);
	giveUp = 1;
	mos_cond_broadcast(&waitCond);
	mos_mutex_unlock(&waitMutex);
}

static void allStatsThread()
{
	for (statsNode = nNodes; statsNode>0; statsNode--)
	{
		uint8_t i;
		mos_mutex_lock(&waitMutex);
		memset(received, 0, sizeof(received));
		giveUp = 0;
		uint8_t retry = 1;
		while (!giveUp) {
			for (i = 0; i<N_STATS_PACKETS; i++)
				if (!received[i])
					break;
			if (i == N_STATS_PACKETS || giveUp)
				break;
			if (retry) {
				printf_P(sQuery, statsNode);
				send_command(DELUGE_COMMAND_SENDSTATS, statsNode, statsPort);
			}
			retry = mos_cond_timedwait(&waitCond, &waitMutex, (uint32_t)12, (uint32_t)0);
		}
		if (giveUp) {
			mos_mutex_unlock(&waitMutex);
			return;
		}
		mos_mutex_unlock(&waitMutex);
		printStats();
	}
}

/*
 * Run an experiment where one node gets reprogrammed.  This command automates
 * the following process:
 * 1. Stop all nodes recording stats.
 * 2. Clear stats on all nodes.
 * 3. Change the version number on one node.
 * 4. Start all nodes recording stats.
 * 5. Wait until one node gets reprogrammed.
 * 6. Query all nodes for stats and neighbor lists.  (Nodes must be numbered 1 
 * to N if there are N nodes in the network.  There should be no gaps.)
 * 
 * Start time, completion time, packet counts, and neighbor lists will be
 * logged to a file, if you start mos_shell with the -l option.
 */
static void runExper()
{
	// Stop recording stats
	putchar('.');	
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');
	// Clear stats
	send_command(DELUGE_COMMAND_CLEARSTATS, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_CLEARSTATS, -1, experPort);
	putchar('\n');	
	
	// Set version on seed node
	mos_mutex_lock(&waitMutex);
	ver_ack = 0;
	giveUp = 0;
	while (!giveUp && !ver_ack) {
		printf_P(sSendVersion, sendVersion, verNode);
		set_version_impl(sendVersion, verNode, experPort);
		mos_cond_timedwait(&waitCond, &waitMutex, (uint32_t)2, (uint32_t)0);
	}
	mos_mutex_unlock(&waitMutex);
	if (giveUp) return;
	
	// Start recording stats now
	putchar('.');	
	send_command(DELUGE_COMMAND_STARTRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_STARTRECORD, -1, experPort);
	putchar('\n');	
	
	// Wait for the receiving node to finish
	mos_mutex_lock(&waitMutex);
	memset(nodeFlags, 0, sizeof(nodeFlags));
	nodeFlags[experReceiver] = RECEIVER | WAIT_FINISH;
	giveUp = 0;
	while (!giveUp && !(nodeFlags[experReceiver] & FINISHED)) {
		mos_cond_wait(&waitCond, &waitMutex);
	}
	mos_mutex_unlock(&waitMutex);
	if (giveUp) return;
	
	// Record for one more minute so we get an accurate count of data packets
	mos_thread_sleep(60000);
	
	// Stop recording
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(1000);
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(1000);
	
	// Query all nodes for stats
	statsPort = experPort;
	allStatsThread();
}

static void exper()
{
	experPort = 2;
	verNode = 25;
	experReceiver = 1;
	printf_P(sVersion);
	sendVersion = prompt_uint8("");
	nNodes = 25;

	mos_thread_new(runExper, 256, PRIORITY_NORMAL);
}

static void runExperDeluge()
{
	// Stop recording stats
	putchar('.');	
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');
	// Clear stats
	send_command(DELUGE_COMMAND_CLEARSTATS, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_CLEARSTATS, -1, experPort);
	putchar('\n');	
	
	// Set version on seed node
	mos_mutex_lock(&waitMutex);
	ver_ack = 0;
	giveUp = 0;
	while (!giveUp && !ver_ack) {
		printf_P(sSendVersion, sendVersion, verNode);
		set_version_impl(sendVersion, verNode, experPort);
		mos_cond_timedwait(&waitCond, &waitMutex, (uint32_t)2, (uint32_t)0);
	}
	mos_mutex_unlock(&waitMutex);
	if (giveUp) return;
	
	// Start recording stats now
	putchar('.');	
	send_command(DELUGE_COMMAND_STARTRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_STARTRECORD, -1, experPort);
	putchar('\n');	
	
	// Wait for the receiving node to finish
	mos_mutex_lock(&waitMutex);
	memset(nodeFlags, WAIT_FINISH, sizeof(nodeFlags));
	nodeFlags[experReceiver] = RECEIVER | WAIT_FINISH;
	giveUp = 0;
	while (!giveUp) {
		uint8_t i;
		for (i=nNodes; i>0; --i)
			if (!(nodeFlags[i] & FINISHED))
				break;
		if (i)
			mos_cond_wait(&waitCond, &waitMutex);
		else
			break;
	}
	mos_mutex_unlock(&waitMutex);
	if (giveUp) return;
	
	// Stop recording
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(1000);
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(1000);
	
	// Query all nodes for stats
	statsPort = experPort;
	allStatsThread();
}

static void exper_del()
{
	experPort = 2;
	verNode = 25;
	experReceiver = 1;
	printf_P(sVersion);
	sendVersion = prompt_uint8("");
	nNodes = 25;

	mos_thread_new(runExperDeluge, 256, PRIORITY_NORMAL);
}

static void runCacheTest()
{
	// Stop recording stats
	putchar('.');	
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');
	// Clear stats
	send_command(DELUGE_COMMAND_CLEARSTATS, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_CLEARSTATS, -1, experPort);
	putchar('\n');	
	
	// Set version on seed node
	mos_mutex_lock(&waitMutex);
	ver_ack = 0;
	giveUp = 0;
	while (!giveUp && !ver_ack) {
		printf_P(sSendVersion, sendVersion, verNode);
		set_version_impl(sendVersion, verNode, experPort);
		mos_cond_timedwait(&waitCond, &waitMutex, (uint32_t)2, (uint32_t)0);
	}
	mos_mutex_unlock(&waitMutex);
	if (giveUp) return;
	
	// Start recording stats now
	putchar('.');	
	send_command(DELUGE_COMMAND_STARTRECORD, -1, experPort);
	mos_thread_sleep(2000);
	putchar('.');	
	send_command(DELUGE_COMMAND_STARTRECORD, -1, experPort);
	putchar('\n');	
	
	// Wait for the receiving node to finish
	mos_mutex_lock(&waitMutex);
	memset(nodeFlags, 0, sizeof(nodeFlags));
	nodeFlags[1] = RECEIVER | WAIT_FINISH;
	nodeFlags[6] = RECEIVER | WAIT_FINISH;
	nodeFlags[11] = RECEIVER | WAIT_FINISH;
	nodeFlags[16] = RECEIVER | WAIT_FINISH;
	nodeFlags[21] = RECEIVER | WAIT_FINISH;
	giveUp = 0;
	while (!giveUp) {
		uint8_t i;
		for (i=1; i<nNodes; i+=5)
			if (!(nodeFlags[i] & FINISHED))
				break;
		if (i < nNodes)
			mos_cond_wait(&waitCond, &waitMutex);
		else
			break;
	}
	mos_mutex_unlock(&waitMutex);
	if (giveUp) return;
	
	// Stop recording
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(1000);
	send_command(DELUGE_COMMAND_ENDRECORD, -1, experPort);
	mos_thread_sleep(1000);
	
	// Query all nodes for stats
	statsPort = experPort;
	allStatsThread();
}

static void cache_test()
{
	experPort = 2;
	cachesize();
	verNode = 25;
	printf_P(sVersion);
	sendVersion = prompt_uint8("");
	nNodes = 25;

	mos_thread_new(runCacheTest, 256, PRIORITY_NORMAL);
}

/*
 * Query all nodes for packet counts.  (Nodes must be numbered 1 
 * to N if there are N nodes in the network.  There should be no gaps.)
 * Packet counts will be logged to a file, if you start mos_shell with 
 * the -l option.
 */
void all_stats()
{
	printf_P(sPort);
	statsPort = prompt_uint8("");
	printf_P(sNNodes);
	nNodes = prompt_long("");
	
	mos_thread_new(allStatsThread, 256, PRIORITY_NORMAL);
}

void start(void)
{
	seq = (uint8_t)rand();
	memset(nodeFlags, 0, sizeof(nodeFlags));
	
	mos_command_daemon_init();
	mos_register_function("send_hello", send_hello);
	//mos_register_function("clear_seq", clear_seq);
	//mos_register_function("stop_all", stop_all);
	mos_register_function("set_version", set_version);
	mos_register_function("clear_stats", clear_stats);
	mos_register_function("start_record", start_record);
	mos_register_function("end_record", end_record);
	mos_register_function("get_stats", get_stats);
	//mos_register_function("get_neighbors", get_neighbors);
	//mos_register_function("wipe", wipe);
	mos_register_function("reboot_net", reboot_net);
	mos_register_function("exper", exper);
	mos_register_function("exper_del", exper_del);
	mos_register_function("all_stats", all_stats);
	mos_register_function("cachesize", cachesize);
	//mos_register_function("cache_test", cache_test);
	mos_register_function("power", power);
	mos_register_function("give_up", give_up);
	
	mos_thread_new(recvThread, 256, PRIORITY_NORMAL);
	mos_thread_new(monitorThread, 256, PRIORITY_NORMAL);
	mos_thread_new(mos_command_daemon, MOS_COMMANDER_STACK_SIZE, PRIORITY_NORMAL);
}