asyncappllayer.cc

基于omnet++开发的Mf框架下的802.11协议仿真。

/*********************************************************
 * file:        AsyncApplLayer.cc
 *
 * author:      Fang shi
 *
 * copyright:   (C) 2006 HUNAN Universtiy, ChangSha China
 *
 ***************************************************************************
 * part of:     Async Simulation
 * description: - Async Description
 *
 ***************************************************************************
 * changelog:   $Revision$
 *              last modified:   $Date: 2007-5-10 10:21 $
 *              by:              $Author: Fang shi $
 ***************************************************************************/


#include "AsyncApplLayer.h"
#include <ApplPkt_m.h>  //该成#include "ApplPkt_m.h"就会出错。
#include <fstream>
#include <string>
#include <cmath>
#include <cstdlib>
#include <algorithm>
#include <iostream>
#include <windows.h>
#include <iterator>
#include <mysql.h>

#pragma comment( lib, "libmysql.lib")
#define EV (ev.disabled()||!debug) ? (std::ostream&)ev : ev << logName() << "::AsyncApplLayer: "

Define_Module(AsyncApplLayer);

MYSQL mysql;
std::deque<std::string> x_delay;//延时统计变量

std::deque<std::string> x_data;//数据变量	
std::deque<std::string> x_rdata;//恢复后数据	
CRITICAL_SECTION m_lock;
//InitializeCriticalSection(&m_lock);

/** 
 * do some initialization 
 * @param stage stage number
 **/
	
std::string int_to_filename(int i,std::string sName="shuju/data")
{
    char str[100];
	  sprintf(str,"%s%d%s",sName.c_str(),i,".txt");
	  std::string s(str);
	  return s;
} 

void AsyncApplLayer::initialize(int stage)
{
	BasicApplLayer::initialize(stage); //DO NOT DELETE!!	
	srand((unsigned)time(NULL));  	
	
	if(stage==0){   
		 int fileNo=findHost()->index();
		 int fileNum=fileNo%56+1;
     //int fileNum=fin dHost()->index();
     if(fileNo!=0)
     {
				std::string file_name=int_to_filename(fileNum);
				std::string textline;
				std::cout<<"file_name = "<<file_name<<endl;
				std::ifstream infile( file_name.c_str(), std::ios::in );
				if (!infile ) {
			   	std::cerr << "oops! unable to open file "
			   		<< file_name << " -- bailing out!\n";
			   	exit( -1 );
			  }
	      while (getline(infile,textline,'\n')) 
	    	   sDeque.push_back(textline);
	      infile.close(); 
	      row=0;
     }
     else
  	 {
	  		system("del /Q copy");
	  		system("del /Q yasuobi.txt");
	  		system("del /Q mse.txt");
	  		system("del /Q result.txt");
	  		system("del /Q delay.txt");
	  		system("del /Q lostpkt.txt");
	  		system("del /Q sinktime.txt");
	  		system("del /Q clustertime.txt");
	  		
	  		/*
	  		**初始化数据库
	  		*/
	  		if(mysql_init(&mysql) == NULL)
	  		{
	  			std::cout<<"init mysql data status fail"<<std::endl;
	  			return;
	  		}
	  		else
	  			std::cout<<"init mysql data status success"<<std::endl;
	  		if(NULL == mysql_real_connect(&mysql,"localhost","root",NULL,"wsnss",MYSQL_PORT,NULL,0))
	  		{
	  			std::cout<<"connect database fail"<<std::endl<<mysql_error(&mysql)<<std::endl;
	  			return;
	  		}	
	  		else
	  			std::cout<<"connect database success"<<std::endl;
	  	  /*
	  	  **初始化数据库结束
	  	  */
  		}
    	pkReceived = 0;
    	numBurstMessage=0;
	    maxNumBurst=(int)par("maxNumBurst"); //max number of bursting message
		
    	WATCH(numBurstMessage);    
      WATCH(pkReceived);

    	destAddress = par("destAddress");    //add sourceAddress	
    	sourceAddress=par("sourceAddress"); //多余   	
    	insertDatabase = new cMessage("insertDatabase");
    	if(parentModule()->index()==destAddress && ev.isGUI())//highlight the sink node
    		parentModule()->displayString().setTagArg("i",1,"yellow");   
    			    	
    
    	     
    	//std::cout<<"aaa"<<std::endl;
    	dataCounter=0;
    	timeStamp=0.0;
    	adr=0;
    	lostCounter=0;
    	mse=0.0;
    	cTopology *topo = new cTopology("topo");
    	topo->extractByModuleType(parentModule()->className(), NULL);    	    	
    	destModuleID=topo->node(destAddress)->module()->findSubmodule("net");// find destAddress node's net module id    	
  		EV << "sink node: " << destAddress << "---net module id is: " << destModuleID <<endl;  				 		  		
  		delete topo;
  		
    	if(parentModule()->index()!=destAddress)
    	{
    		generatePacket = new cMessage("generatePacket");
    		scheduleAt(simTime()+(double)par("iaitime"), generatePacket);    ////    		
    	}
    	else 
    	{
    	  scheduleAt(simTime()+5.0,insertDatabase);
    	}
    			
    //	stopptr=-2;
    	
    	EV<<" AsyncApplLayer is loaded!!!"<<endl;    	
  }
/*  else if(stage==1){
		
  }*/
}


/** 
 * You got a message from the network layer (most likely)
 * take care of it
 * @param msg msg to handle
 **/
void AsyncApplLayer::handleLowerMsg( ApplPkt* msg )
{   	
  	 pkReceived++;   		
  	 char str1[1024];
		 sprintf(str1,"insert into delay_n(ID,simTime,creatTime,delayTime)values('%d','%lf','%lf','%lf');",
		 findHost()->index(),simTime(),msg->timestamp(),simTime()-msg->timestamp());
		 			
		 std::string insert(str1);
		 x_delay.push_back(insert);
  	 double temp1[8];
  	 for(int i=0;i<8;i++)
  	 {
  	 	temp1[i]=msg->getData(i);
     }
         //插入数据库
     char str[1024];
		 sprintf(str,"insert into data_receive_n(ID,data1,data2,data3,data4,data5,data6,data7,data8)values('%d','%lf','%lf','%lf','%lf','%lf','%lf','%lf','%lf');",
			 msg->getSrcAddr(),temp1[0],temp1[1],temp1[2],temp1[3],temp1[4],temp1[5],temp1[6],temp1[7]);
			
	   std::string insert1 (str);
		 x_rdata.push_back(insert1);
     
     delete msg;
     if (ev.isGUI())
       	parentModule()->bubble("Arrived!");   
      
     displayStatus(true);       
	 
}

void AsyncApplLayer::displayStatus(bool isBusy)
{
    char string[50];
    if(findHost()->index()==(int)findHost()->submodule("appl")->par("destAddress"))
    {
    	sprintf( string, "%d", pkReceived);    
      findHost()->displayString().setTagArg("t",0, string);    
    }
}


/** 
 * You have send yourself a message -- probably a timer,
 * take care of it
 * @param msg msg to handle
 **/
void AsyncApplLayer::handleSelfMsg(cMessage *msg){
    
    if(msg==generatePacket && numBurstMessage<maxNumBurst)
    {     	             
      ApplPkt *pk = new ApplPkt();
	    std::string s=sDeque[row];
	    	   	
	  	std::string::size_type pos = 0, prev_pos = 0;
	  	unsigned int index=0;
	  	double data1[8];
	  	double data;
	  	while (( pos = s.find_first_of( ' ', pos ))!= std::string::npos )
	  	{
	   		//++pos;
       	//pos = s.find_first_of(' ',pos);
	      if(s.substr(prev_pos, pos-prev_pos)=="")//如果等于空字符串，继续查找
	      {
	      	prev_pos=++pos;
	        continue;
        } 
        else
        {
        	data = atof(s.substr(prev_pos, pos - prev_pos).c_str());//?
        	data1[index++]=data;
        	prev_pos = ++pos;
        }	
      }	  
			data=atof(s.substr(prev_pos,pos - prev_pos).c_str());				 
			data1[index]=data;
  			
	  /*将数据包真实数据插入数据库 */
	    char str[1024];
			sprintf(str,"insert into data_n(ID,data1,data2,data3,data4,data5,data6,data7,data8)values('%d','%lf','%lf','%lf','%lf','%lf','%lf','%lf','%lf');",
					findHost()->index(),data1[0],data1[1],data1[2],data1[3],data1[4],data1[5],data1[6],data1[7]);
				
			std::string insert(str);
			x_data.push_back(insert);
			  
    
			for(int i=0;i<8;i++)
			{ 
				ev<<"*"<<endl;
				pk->setData(i,data1[i]);				
			}
				
	/**###################################################*/
	               /** 数据报设置*/
	     
       //pk->setSrcAddr(myApplAddr());
       pk->setSrcAddr(findHost()->index());
       pk->setDestAddr(destModuleID);
       pk->setLength(headerLength);
       pk->setTimestamp(simTime()); 
       pk->setSeqNum1(row);//4-12添加
       row++;
      // std::cout<<"aaa"<<std::endl;
       ev<<logName()<<"::AsyncApplLayer:: the pk's timestamp=="<<pk->timestamp()<<endl;  
       
       sendDown(pk);
	
		   if (ev.isGUI()) 
		    	parentModule()->bubble("Generating packet...");    	
       numBurstMessage++;
       EV <<" generating packet to " <<destAddress<< "#" << numBurstMessage <<endl;
       scheduleAt(simTime()+(double)par("iaitime"), generatePacket);          
    }  
    else if(msg==insertDatabase&&findHost()->index()==(int)findHost()->submodule("appl")->par("destAddress"))  //插入数据库,在sink节点才进行操作。
    {
    	 //::EnterCriticalSection(&m_lock);
//    	 std::cout<<"insert into database"<<std::endl;
//    	 std::cout<<x_delay.size()<<std::endl;
    	 		 
			 
			 //出入传感数据
			 for(std::deque<std::string >::iterator iter = x_data.begin();iter != x_data.end();++iter)
			 {
			 		if(mysql_query(&mysql,(*iter).c_str())!=0)
			 		{
			// 			std::cout<<"execute insert syntax fail"<<
				//				std::endl<<mysql_error(&mysql)<<endl;
						mysql_close(&mysql);
						return;
			 		}
			 }
			 x_data.clear();
			 
			 for(std::deque<std::string >::iterator iter = x_delay.begin();iter != x_delay.end();++iter)
			 {
			 		if(mysql_query(&mysql,(*iter).c_str())!=0)
			 		{
			// 			std::cout<<"execute insert syntax fail"<<
			//					std::endl<<mysql_error(&mysql)<<endl;
						mysql_close(&mysql);
						return;
			 		}
			 }
			 x_delay.clear();
			 
			  for(std::deque<std::string >::iterator iter = x_rdata.begin();iter != x_rdata.end();++iter)
			 {
			 		if(mysql_query(&mysql,(*iter).c_str())!=0)
			 		{
			// 			std::cout<<"execute insert1 syntax fail"<<
				//				std::endl<<mysql_error(&mysql)<<endl;
						mysql_close(&mysql);
						return;
			 		}
			 }
			 x_rdata.clear();
			 
					 
			 scheduleAt(simTime()+5.0,insertDatabase);
			

    }
    
    
    else if(numBurstMessage>=maxNumBurst)
   		EV << " source stopped bursting!!! " << endl;   		
   	
}


void AsyncApplLayer::finish()
{

	m_iterator it=dmap.begin();
	m_iterator end=dmap.end();
	for(;it!=end;++it)
	{
		//std::cout<<"cluster head's address is: "<<(*it).first<<", the received packet number is: "<<(*it).second.size()<<std::endl;
	}
}