deluge_app.c

来自「MANTIS是由科罗拉多大学开发的传感器网络嵌入式操作系统。 这是mantis」· C语言 代码 · 共 1,317 行 · 第 1/3 页

		{
			if ((txpackets[txidx] & txmask)!=0)
			{
				mos_mutex_lock(&app->delugeLock);
				mos_file_read(spkt.data, app->image_file, txaddr, DELUGE_PACKET_SIZE);
				mos_mutex_unlock(&app->delugeLock);
							
				foot->packet = i;
				spkt.size = DELUGE_PACKET_SIZE+sizeof(deluge_foot_data);	// com_send may have changed it
							
				// Last two bytes of last packet have page CRC
				if (i == DELUGE_PACKETS_PER_PAGE-1) {
					*(uint16_t*)&(spkt.data[DELUGE_PACKET_SIZE-2]) = crc;
				}
							
				net_send(&spkt, DELUGE_PROTO_ID, deluge_portmap[app->index], deluge_portmap[app->index]);
			
				ih = mos_disable_ints();
				app->outgoingPackets[txidx] &= ~txmask;
				txpackets[txidx] = app->outgoingPackets[txidx];
				mos_enable_ints(ih);
				
				mos_thread_sleep(app->txDelay);
			}
			
			txaddr = txaddr + DELUGE_PACKET_SIZE;
			txmask = txmask << 1;
			if (txmask == 0) {
				txmask = 1;
				txidx++;
			}
		}
				
		ih = mos_disable_ints();
		txpackets[0] = app->outgoingPackets[0];
		txpackets[1] = app->outgoingPackets[1];
		txpackets[2] = app->outgoingPackets[2];
		mos_enable_ints(ih);
	}
	// Rule T.2
	//app->state = DELUGE_STATE_MAINTAIN;
	//mos_sem_post(&timeout_sem);
	//deluge_wakeup(&table[app->index]);
	stateTransition(app, DELUGE_STATE_MAINTAIN);
}

/* FORWARD state is a combination of RX and TX states.  The node will fill as
 * many requests as it can from its cache, and forward requests it can't handle
 * to its parent.  The node's state will return to MAINTAIN once the request
 * is fully filled or too many requests time out.  To minimize contention, data
 * is only forwarded when the page is completed or the request times out or a
 * new request is received.
 */
static void doForward(deluge_app* app)
{
	// TX
	deluge_foot_data* foot = (deluge_foot_data*)&spkt.data[DELUGE_PACKET_SIZE];
	mos_mutex_lock(&app->delugeLock);
	foot->version = app->dcb.version;
	#if DELUGE_CACHE_PAGES > 1
	int8_t page = app->outgoingCachePage;
	#else
	int8_t page = 0;
	#endif
	foot->page = app->outgoingPage;
	foot->id = mos_node_id_get();
	foot->type = DELUGE_PACKET_DATA;

	// Packets that have been requested of us
	uint16_t txpackets[3] = { app->outgoingPackets[0], app->outgoingPackets[1], app->outgoingPackets[2] };
	// Packets we have cached
	uint16_t capackets[3] = { app->cachedPackets[0], app->cachedPackets[1], app->cachedPackets[2] };
	mos_mutex_unlock(&app->delugeLock);

	app->nPacketsRequested = 0;
	// Loop as long as we have packets to fill the request
	while (app->state == DELUGE_STATE_FORWARD &&
		((txpackets[0]&capackets[0]) != 0 || (txpackets[1]&capackets[1]) != 0 || (txpackets[2]&capackets[2]) != 0))
	{
		uint8_t i, ih;
		uint16_t txmask = 1;
		uint8_t txidx = 0;
		uint32_t txaddr = DELUGE_ADDR(page, 0);
		for (i=0; i<DELUGE_PACKETS_PER_PAGE; i++)
		{
			if ((txpackets[txidx] & capackets[txidx] & txmask)!=0)
			{
				mos_mutex_lock(&app->delugeLock);
				mos_file_read(spkt.data, app->image_file, txaddr, DELUGE_PACKET_SIZE);
				mos_mutex_unlock(&app->delugeLock);
							
				foot->packet = i;
				spkt.size = DELUGE_PACKET_SIZE+sizeof(deluge_foot_data);	// net_send may have changed it
							
				// Last two bytes of last packet have page CRC
				// We don't compute it ourselves, since we might not have
				// the entire page cached
				if (i == DELUGE_PACKETS_PER_PAGE-1) {
					*(uint16_t*)&(spkt.data[DELUGE_PACKET_SIZE-2]) = app->incomingcrc;
				}
							
				net_send(&spkt, DELUGE_PROTO_ID, deluge_portmap[app->index], deluge_portmap[app->index]);
			
				ih = mos_disable_ints();
				app->outgoingPackets[txidx] &= ~txmask;
				txpackets[txidx] = app->outgoingPackets[txidx];
				mos_enable_ints(ih);
				
				mos_thread_sleep(app->txDelay);
			} else if ((txpackets[txidx] & txmask) != 0) {
				app->nPacketsRequested++;
			}
			
			txaddr = txaddr + DELUGE_PACKET_SIZE;
			txmask = txmask << 1;
			if (txmask == 0) {
				txmask = 1;
				txidx++;
			}
		}
				
		ih = mos_disable_ints();
		txpackets[0] = app->outgoingPackets[0];
		txpackets[1] = app->outgoingPackets[1];
		txpackets[2] = app->outgoingPackets[2];
		capackets[0] = app->cachedPackets[0];
		capackets[1] = app->cachedPackets[1];
		capackets[2] = app->cachedPackets[2];
		
		// As long as we are filling requests, don't let incoming requests
		// inflate our total timeout count
		if (app->nRequests) app->nRequests--;
		mos_enable_ints(ih);
	}
	
	if (txpackets[0]==0 && txpackets[1]==0 && txpackets[2]==0) {
		// We are done
		app->nRequests = 0;
		//app->state = DELUGE_STATE_MAINTAIN;
		//deluge_wakeup(&table[app->index]);
		stateTransition(app, DELUGE_STATE_MAINTAIN);
		return;
	}

	// RX
	if (app->nRequests < DELUGE_LAMBDA) {
		// Rule R.1
		deluge_foot_request* request = (deluge_foot_request*)spkt.data;
		mos_mutex_lock(&app->delugeLock);
		request->to = app->updaterNode;
		// Request the packets that were requested of us
		request->packets[0] = (app->incomingPackets[0] |= app->outgoingPackets[0]);
		request->packets[1] = (app->incomingPackets[1] |= app->outgoingPackets[1]);
		request->packets[2] = (app->incomingPackets[2] |= app->outgoingPackets[2]);
		request->version = app->dcb.version;
		request->page = app->dcb.incomingPage;
		mos_mutex_unlock(&app->delugeLock);
		request->rateChange = app->nRequests-1;

		request->id = mos_node_id_get();
		request->type = DELUGE_PACKET_REQUEST;
		spkt.size = sizeof(deluge_foot_request);
					
		net_send(&spkt, DELUGE_PROTO_ID, deluge_portmap[app->index], deluge_portmap[app->index]);
		app->nRequests++;
	} else {
		// TODO: check reception rate
		app->nRequests = 0;
		// Assume parent node is dead; assign the highest possible DTB; 
		// we'll take any node with a lower DTB
		app->dtb = 0xFF;
		/*#ifdef DELUGE_LINK_QUALITY
		app->dtb_votes[app->updater_index] = 0;
		#endif*/
		#ifdef DELUGE_SYMMETRIC_LINKS
		clearSymmetry(app, app->updaterNode);
		/*if (nextSymmetric(app) != 255) {
			app->nRequests = 0;
			stateTransition(app, DELUGE_STATE_FORWARD);
			return;
		}*/
		#endif
		#ifdef DELUGE_CLEAR_OUTGOING
		app->outgoingPackets[0] = 0;
		app->outgoingPackets[1] = 1;
		app->outgoingPackets[2] = 2;
		#endif
		stateTransition(app, DELUGE_STATE_MAINTAIN);
		return;
	}

	// wait a while before requesting again
	setRequestTimeout(app, DELUGE_STATE_FORWARD);
}

void deluge_dispatch(deluge_app* app)
{
	/*if (app->wakeup_reason == PACKET_RECEIVED)
		clearTimeout(app);*/
	switch (app->state)
	{
		case DELUGE_STATE_MAINTAIN:
			doMaintain1(app);
			break;
		case DELUGE_STATE_MAINTAIN2:
			doMaintain2(app);
			break;
		case DELUGE_STATE_RX:
			doRx(app);
			break;
		case DELUGE_STATE_TX:
			doTx(app);
			break;
		case DELUGE_STATE_FORWARD:
			doForward(app);
			break;
	}
}

void handleSummary(deluge_app* app, deluge_foot_summary* summary)
{
	mos_mutex_lock(&app->delugeLock);
	checkNeighbor(app, summary->id);
	mos_mutex_unlock(&app->delugeLock);
	
	switch (app->state)
	{
		case DELUGE_STATE_MAINTAIN:
		case DELUGE_STATE_MAINTAIN2: {
			mos_mutex_lock(&app->delugeLock);
			// Nodes that are far away from the base should not suppress us
			//if (summary->dtb < app->dtb)
				app->nSummaries++;
			if (
				#ifndef DELUGE_NO_FORWARD
				app->index == 0 &&
				#endif
				#ifdef DELUGE_SYMMETRIC_LINKS
				checkSymmetry(app, summary) &&
				#endif
				summary->version == app->dcb.version &&
				summary->highPage > app->dcb.highPage
				#ifndef DELUGE_NO_FORWARD
				#ifndef MEMBER_NO_DTB
				&& summary->dtb < app->dtb
				#else
				&& summary->dtb < 255
				#endif
				#endif
				)
			{
				// We ignore summaries from a node that is farther from the root
				// than our last parent.  If this parent ever fails to update
				// us, we set the DTB to the maximum, in which case we no longer
				// ignore summaries.
				app->updaterNode = summary->id;
				// DTB ensures that we get our updates from the closest source
				app->dtb = summary->dtb + 1;
				// Rule M.5
				if (!app->heardRequest && !app->heardData) {
					app->nRequests = 0;
					stateTransition(app, DELUGE_STATE_RX);
				}
				app->detectedInconsistency = 1;
			}
			#ifndef DELUGE_NO_FORWARD
			else if (app->index != 0 &&
				#ifdef DELUGE_SYMMETRIC_LINKS
				checkSymmetry(app, summary) &&
				#endif
				summary->version == app->dcb.version &&
				(summary->highPage > app->dcb.highPage ||
				(summary->highPage == app->dcb.highPage &&
				/*#ifdef DELUGE_LINK_QUALITY
				checkDtb(app, summary->id, summary->dtb)
				#else*/
				summary->dtb < app->dtb
				//#endif
				)))
			{
				// We will send this page index in our own summaries
				app->dcb.highPage = summary->highPage;
				// DTB ensures that we get our updates from the closest source
				// and prevents routing loops from forming
				app->dtb = summary->dtb + 1;
				app->detectedInconsistency = 1;
				// Take note of our new parent, but don't switch to RX state
				app->updaterNode = summary->id;
				// If we have a previously pending request, switch back to
				// FORWARD state
				// Rule M.5
				if (!app->heardRequest && !app->heardData &&
					((app->incomingPackets[0]&app->outgoingPackets[0]) ||
					(app->incomingPackets[1]&app->outgoingPackets[1]) ||
					(app->incomingPackets[2]&app->outgoingPackets[2])))
				{
					app->nRequests = 0;
					stateTransition(app, DELUGE_STATE_FORWARD);
				}
			}
			#endif
			else if (summary->version != app->dcb.version ||
				summary->highPage != app->dcb.highPage ||
				(summary->highPage == app->dcb.highPage &&
				summary->dtb > app->dtb+1))
			{
				app->detectedInconsistency = 1;
				if (summary->version < app->dcb.version)
					app->heardObsolete = 1;
				stateTransition(app, DELUGE_STATE_MAINTAIN);
			}
			mos_mutex_unlock(&app->delugeLock);
		} break;
		case DELUGE_STATE_RX: break;
		case DELUGE_STATE_TX: break;
		case DELUGE_STATE_FORWARD: break;
	}
}

void handleProfile(deluge_app* app, deluge_foot_profile* profile)
{
	mos_mutex_lock(&app->delugeLock);
	checkNeighbor(app, profile->id);
	mos_mutex_unlock(&app->delugeLock);
	
	switch (app->state)
	{
		case DELUGE_STATE_MAINTAIN:
		case DELUGE_STATE_MAINTAIN2: {
			mos_mutex_lock(&app->delugeLock);
			app->nProfiles++;
			if (profile->version > app->dcb.version)
			{
				#ifdef DELUGE_PRINT_EVENT
				printf_P(sNewVer);
				printf_P(sDisplayTime);		// display time
				#endif
				aqshell_send(AQSHELL_NEWVERSION, &profile->version, 1);
				
				app->dcb.codeSize = profile->codeSize;
				app->dcb.version = profile->version;
				app->dcb.incomingPage = 0;
				app->dcb.highPage = 0;
				app->dcb.goalPage = profile->goalPage;
				app->dcb.programcrc = profile->crc;
				app->outgoingPackets[0] = 0;
				app->outgoingPackets[1] = 0;
				app->outgoingPackets[2] = 0;
				app->dtb = 0xFF;
				
				uint8_t i;
				for (i=0; i<sizeof(app->dcb.pagesNeeded); i++)
					app->dcb.pagesNeeded[i] = 0xFF;

				// Find the next page after the last program image we wrote
				if (app->index == 0) {
					app->image_file = mos_file_create("prg", app->dcb.codeSize);
					app->incomingPackets[0] = -1;
					app->incomingPackets[1] = -1;
					app->incomingPackets[2] = -1;
					app->nPacketsRequested = DELUGE_PACKETS_PER_PAGE;
				} else {
					char name[4] = { 'a', 'q', '0'+app->index, 0 };
					#ifdef DELUGE_NO_FORWARD
					app->image_file = mos_file_create(name, app->dcb.codeSize);
					app->incomingPackets[0] = -1;
					app->incomingPackets[1] = -1;
					app->incomingPackets[2] = -1;
					app->nPacketsRequested = DELUGE_PACKETS_PER_PAGE;
					#else
					app->image_file = mos_file_create(name, 
						(uint32_t)DELUGE_PAGE_SIZE*(uint32_t)DELUGE_CACHE_PAGES);
					app->incomingPackets[0] = 0;
					app->incomingPackets[1] = 0;
					app->incomingPackets[2] = 0;
					app->nPacketsRequested = 0;
					#endif
					#if DELUGE_CACHE_PAGES > 1
					int8_t i;
					for (i=0; i<DELUGE_CACHE_PAGES; i++)
						app->cache_map[i] = -1;
					app->outgoingCachePage = 0;
					app->incomingCachePage = 0;
					#endif
					app->cachedPackets[0] = 0;
					app->cachedPackets[1] = 0;
					app->cachedPackets[2] = 0;
				}
				deluge_saveState(app);
			}
			mos_mutex_unlock(&app->delugeLock);
		} break;
		case DELUGE_STATE_RX: break;
		case DELUGE_STATE_TX: break;
		case DELUGE_STATE_FORWARD: break;
	}
}

void handleRequest(deluge_app* app, deluge_foot_request* request)
{
	mos_mutex_lock(&app->delugeLock);
	checkNeighbor(app, request->id);
	mos_mutex_unlock(&app->delugeLock);
	
	switch (app->state)
	{
		case DELUGE_STATE_MAINTAIN:
		case DELUGE_STATE_MAINTAIN2: {
			mos_mutex_lock(&app->delugeLock);
			app->detectedInconsistency = 1;
			if (request->version == app->dcb.version &&
				request->page < app->dcb.highPage)
			{
				if (request->to == mos_node_id_get())
				{
					// Rule M.6
					app->outgoingPage = request->page;
					uint8_t ih = mos_disable_ints();
					app->outgoingPackets[0] = request->packets[0];
					app->outgoingPackets[1] = request->packets[1];
					app->outgoingPackets[2] = request->packets[2];
					mos_enable_ints(ih);
					
					app->txDelay += request->rateChange * DELUGE_TX_DELAY_INC;
					if (app->txDelay < DELUGE_TX_DELAY_MIN) 
						app->txDelay = DELUGE_TX_DELAY_MIN;
					else if (app->txDelay > DELUGE_TX_DELAY_MAX)
						app->txDelay = DELUGE_TX_DELAY_MAX;
					
					#ifdef DELUGE_NO_FORWARD
					stateTransition(app, DELUGE_STATE_TX);
					#else
					if (app->index == 0) {
			 			stateTransition(app, DELUGE_STATE_TX);
					} else if (app->dtb != 255) {
						#if DELUGE_CACHE_PAGES > 1
						// Is the page in the cache?
						int8_t page = cachedPageIndex(app, request->page);
						if (page >= 0) {	// Yes