PEtherApp Class Reference

#include <PEtherApp.h>

Inheritance diagram for PEtherApp:

List of all members.

Public Member Functions
virtual void	setNewFID (InformationItem *II, string FID, QoSList qos, vector< string > lidpath)
Protected Types
typedef std::map < InformationItem *, StreamData *, InformationItem >	FIDMap
	The module's FID Table.
typedef FIDMap::iterator	FIDIter
	The iterator for the FID Table.
Protected Member Functions
virtual void	initialize (int stage)
virtual int	numInitStages () const
virtual void	handleMessage (cMessage *msg)
virtual void	finish ()
virtual MACAddress	resolveDestMACAddress ()
virtual void	chooseSubscription ()
virtual void	cancelInformationItem ()
virtual void	sendStreamData (cMessage *timer)
virtual void	receivePacket (cMessage *msg)
virtual bool	genInformationItem ()
Protected Attributes
FIDMap	FID_map
double	psProportion
	Proportion between the publishers and subscribers.
MACAddress	destMACAddress
	The publication MAC address -- Preferably Broadcast.

---

Detailed Description

The Publish Ethernet Application. This module acts more like a streaming application. The only interesting this about it, is the use of FID for the proper routing of the created streams, from the publisher to all assigned subscribers. The FID assignment is a operation of the Topology Manager, and this is why, the function setNewFID() is made public, so that the Topology Manager module of the simulation will be able to call it.

---

Member Typedef Documentation

typedef FIDMap::iterator PEtherApp::FIDIter [protected]

The iterator for the FID Table.

typedef std::map<InformationItem *, StreamData *, InformationItem> PEtherApp::FIDMap [protected]

The module's FID Table.

---

Member Function Documentation

void PEtherApp::cancelInformationItem

(

)

[protected, virtual]

Cancels the publication of a random Information Item from this application. The difference between this publisher and the base application BasePSApp, in this function is that the publisher has also to delete the chosen Information Item element form the FID Table as well.

                                      {

        //If element was successfully removed from the Information Item table, remove it
        //from the FID Table as well
        if(BasePSApp::cancelInformationItem(UNPUBLISH_INFO)) {

                ev << this->getParentModule()->getFullName() << ": Canceling publication." << endl;

                //For all entries in the FID Table of this module find the one missing from the
                //subscriptions table and delete it
                for(FIDIter iter=FID_map.begin();iter!=FID_map.end();iter++) {

                        subscriptionItem subitem;
                        subitem.ii=iter->first;
                        subscrIter subiter = find(subscriptions.begin(),subscriptions.end(), subitem);

                        if(subiter==subscriptions.end()) {
                                ev << "Deleting Information item from FID Table: " << iter->first << endl;
                                //Cancel Timer self-message
                                iter->second->tmsg=cancelEvent(iter->second->tmsg);

                                //Delete Timer self-message
                                delete iter->second->tmsg;

                                FID_map.erase(iter);
                        }
                }
        }
}

void PEtherApp::chooseSubscription

(

)

[protected, virtual]

Chooses a random subscription from the Information Item Table, according to the popularity of each element. Afterwards, the application module tries to become a publisher to this subscription, although it might be already a publisher of it.

                                   {

        //Try to receive a subscriber request for an Information Item
        InformationItem *subscription=returnRandII();
        if(subscription!=NULL) {

                //Inform the Global Information Item Table
                if(iit->addPubElement(subscription,this->getParentModule())) {

                        ev << this->getParentModule()->getFullName() << ": Existing Information Item met with new publication!" << endl;

                        subscriptionItem subitem;
                        subitem.ii=subscription;
                        subscriptions.push_back(subitem);
                }
        }
        //If non returned for any reason print an error message
        else {
                ev << this->getParentModule()->getFullName() << ": Tried to find subscription but list was empty!" << endl;
        }
}

void PEtherApp::finish

(

)

[protected, virtual]

The finish function of the application. Just calls the finish function of the base class and deletes the FID Table of this publisher application.

Reimplemented from BasePSApp.

{
        //Delete all inactive timers of FID_map
        for(FIDIter iter=FID_map.begin();iter!=FID_map.end();iter++) {
        if(iter->second->active==false) {
                        //These items are not scheduled since they are paused
                        if(iter->second->tmsg!=NULL)
                                delete iter->second->tmsg;
                }
        }

        //Clear the FID table
        FID_map.clear();

        //Call the base function finish function
        BasePSApp::finish();
}

bool PEtherApp::genInformationItem

(

)

[protected, virtual]

It generates a information Item and places it into the subscriptions table of the publisher, and informs the Information Item Table module, if exists.

                                   {
        //Create subscription item
        subscriptionItem subitem;

        //Generate random Information Item ID
        char ID[PURSUIT_ID_LEN+1];
        for(int i=0;i<PURSUIT_ID_LEN;i++)
        {
                        ID[i]=(char)(1+intrand(255));
        }
        //Put string end character
        ID[PURSUIT_ID_LEN] = '\0';

        //Generate a scope ID -- static is just fine
        InformationItem *II=new InformationItem("0457A0B220ABF100",ID);

        //Assign the successfully created Information Item in the subscription
        subitem.ii=II;

        //Create randomly Meta-data for the future Information Item
        //TODO: Find a method for generating these metadata
        subitem.qos[QoS_BW_M]=(uint16_t)datarate->doubleValue();
        subitem.qos[QoS_FLOW]=1;

        /* First the Topology Manager has to be informed of the new Information Item
         * and it's meta data. This happens to avoid finding paths without being
         * informed of the QoS requirements of each item. If there is a mistake,
         * such as the Information Item already exists, the Topology Manager will
         * not accept the packet. The same information lies on the rendezvous table.*/
        publishMetaDataPacket(subitem);

        /***Inform the Global Information Item Table***/
        if(iit->addPubElement(II,this->getParentModule())) {

                subscriptions.push_back(subitem);
                return true;
        }
        /**********************************************/

        return false;
}

void PEtherApp::handleMessage

(

cMessage *

msg

)

[protected, virtual]

Handles incoming messages for the publisher application. In case of incoming messages, it records some statistics, although this functionality is just kept for future use. The most important use of this function is the interpretation of the received self messages, controlling the transmission of packets for the published Information Item streams and the creation new publications.

Reimplemented from BasePSApp.

{
    if (msg->isSelfMessage())
    {
        //Self message to create or delete a new publication
        if(strcmp(msg->getFullName(),"chooseSubscriber")==0) {
                //Decide whether to make new subscription
                        double now=simTime().dbl();
                        //Get a random variable according the the Gaussian distribution
                        double pubPeak=normal(this->par("actPeak").doubleValue(),PEAK_DEV);

                        //Try to create a new publication, if the random variable is close enough to the current time.
                        //To create a periodical functionality, the modulo of the current time is taken for the time
                        //period equal to three times the mean value of the Gaussian random variable.
                        if(pubPeak-(2*PEAK_DEV+1)<fmod(now,3*pubPeak) && pubPeak+(2*PEAK_DEV+1)>fmod(now,3*pubPeak)) {
                                if(dblrand()<0.8) {
                                        chooseSubscription();
                                }
                                else {
                                        genInformationItem();
                                }
                }
                        //Or delete an old one if the current time is close enough the the double of the Gaussian mean.
                        //Notice that a bit more space is given to the removing of publications, since it creates
                        //better results.
                        else if(2*pubPeak-(2*PEAK_DEV+2)<fmod(now,3*pubPeak) && 2*pubPeak+(2*PEAK_DEV+2)>fmod(now,3*pubPeak)) {
                        cancelInformationItem();
                }

                simtime_t pt = par("pubInterval").doubleValue();
                // Send packet with a small jitter
                simtime_t jitter = dblrand() * MAXJITTER;
                scheduleAt(simTime()+pt+jitter, msg);
        }
        //Self message indicating it is time to send a stream
        else if(strcmp(msg->getFullName(),"StreamTmr")==0) {
                sendStreamData(msg);
        }
    }
    else
    {
        receivePacket(msg);
    }

}

void PEtherApp::initialize

(

int

stage

)

[protected, virtual]

Initialization function of the publisher application. Schedules the initial self messages and the class parameters.

Reimplemented from BasePSApp.

{
        BasePSApp::initialize(stage);

    // we can only initialize in the 2nd stage (stage==1), because
    // assignment of "auto" MAC addresses takes place in stage 0
    if (stage == 1)
    {
        //Read the proportion of choosing publishers against adding subscribers
        psProportion = simulation.getSystemModule()->par("PubSubProportion");
                if(psProportion<0 || psProportion>1)
                        psProportion=0.5;

        //Watch the FID Table
        WATCH_MAP(FID_map);

        destMACAddress = resolveDestMACAddress();

        // if no dest address given, nothing to do
        if (destMACAddress.isUnspecified())
            return;

        //Generate self message for choosing a subscriber to publish for
        cMessage *chooseSubmsg = new cMessage("chooseSubscriber");
                simtime_t pt = par("pubInterval").doubleValue();
                // Send packet with a small jitter
                simtime_t jitter = dblrand() * MAXJITTER;
                scheduleAt(simTime()+pt+jitter, chooseSubmsg);
    }
}

virtual int PEtherApp::numInitStages

(

)

const [inline, protected, virtual]

Returns the number of the initialization stages.

Reimplemented from BasePSApp.

{return 5;}

void PEtherApp::receivePacket

(

cMessage *

msg

)

[protected, virtual]

It actually only records their statistics and deletes them. It has now current use and it is kept for future expansion of the class.

{
    EV << "Received packet `" << msg->getName() << "'\n";

    packetsReceived++;
    simtime_t lastEED = simTime() - msg->getCreationTime();
    eedVector.record(lastEED);
    eedStats.collect(lastEED);

    delete msg;
}

MACAddress PEtherApp::resolveDestMACAddress

(

)

[protected, virtual]

Interprets the MAC address chosen as in input parameter for sending all publication packets to.

{
    MACAddress destMACAddress;
    const char *destAddress = par("destAddress");
    if (destAddress[0])
    {
        // try as mac address first, then as a module
        if (!destMACAddress.tryParse(destAddress))
        {
            cModule *destStation = simulation.getModuleByPath(destAddress);
            if (!destStation)
                error("cannot resolve MAC address '%s': not a 12-hex-digit MAC address or a valid module path name", destAddress);
            cModule *destMAC = destStation->getSubmodule("mac");
            if (!destMAC)
                error("module '%s' has no 'mac' submodule", destAddress);
            destMACAddress.setAddress(destMAC->par("address"));
        }
    }
    return destMACAddress;
}

void PEtherApp::sendStreamData

(

cMessage *

timer

)

[protected, virtual]

Send the packets of a stream corresponding to a specific Information Item, in the interval determined by the waitInterval variable of the class. The packets are sent to the MAC address specified by the destMACAddress variable of the class and with FID taken from the FID Table of the application. In case the FID is empty, no packet is sent until a new value is filled in.

{
        //Access current stream
        InformationItem *II = (InformationItem *) timer->getContextPointer();

        FIDIter iter=FID_map.find(II);

        if(iter==FID_map.end()) {
                ev << this->getParentModule()->getFullName() << ": Error trying to send a deleted Information Item stream!" << endl;

                //Delete obsolete timer
                delete timer;
        }
        else {

                //If FID is not empty, continue sending stream
                if(iter->second->FID.compare("")!=0) {

                        seqNum++;

                        // generate and send a packet
                        char msgname[30];
                        sprintf(msgname, "PSmsg-%d-%ld", getId(), seqNum);
                        ev << "Generating packet `" << msgname << "'\n";

                        PSEtherAppMsg *datapacket = new PSEtherAppMsg(msgname, IEEE802CTRL_DATA);

                        // ---------------------------------------------//
                        // ------------- Set Packet header -------------//
                        // ---------------------------------------------//

                        datapacket->setFID(iter->second->FID.c_str());

                        //Set the information item scope length
                        datapacket->setScopeLen(iter->first->scopeSize());

                        //Set the Information Item scope and ID
                        datapacket->setIIID(*iter->first);

                        //Set the information item type
                        datapacket->setType(PUBLISHED_DATA);

                        //If we are checking for false positives. This field does not count in the
                        //bit length of the message since it is supposed not to be there!
                        if(simulation.getSystemModule()->par("FP_check").boolValue()==true) {
                                //Set the path length
                                datapacket->setPathArraySize(iter->second->path.size());
                                //Set the lid of each link
                                for(uint i=0;i<iter->second->path.size();i++)
                                        datapacket->setPath(i,iter->second->path[i].c_str());
                        }

                        Ieee802Ctrl *etherctrl = new Ieee802Ctrl();
                        etherctrl->setDest(destMACAddress);
                        datapacket->setControlInfo(etherctrl);

                        //Fix the size of the packet
                        int length=datapacket->getByteLength();
                        datapacket->setByteLength(length + FID_LEN /*FID*/ + 1 /*type*/ + 1 /*scopeLen*/ + PURSUIT_ID_LEN*iter->first->scopeSize() /*Scope*/ + PURSUIT_ID_LEN /*Inf_Item ID*/);

                        //---------------------------------------------//
                        //-------------- Set Packet Load --------------//
                        //---------------------------------------------//
                        datapacket->setByteLength(MAX_ETH_MTU);
                        //Check if Information Item represents stream or a file
                        /*if(iter->second->byteSize>=0) {
                                //If file and the remaining size is more than the maximum remaining payload
                                if(iter->second->byteSize>(MAX_ETH_MTU-length)) {
                                        datapacket->setByteLength(MAX_ETH_MTU);
                                        iter->second->byteSize-=(MAX_ETH_MTU-length);
                                }
                                //Else sent all the remaining bytes
                                else {
                                        int length=datapacket->getByteLength();
                                        datapacket->setByteLength(length + iter->second->byteSize);
                                        iter->second->byteSize=0;
                                }
                        }
                        //If stream leave a byte free for differentiation in the reception
                        else
                                datapacket->setByteLength(MAX_ETH_MTU-1);*/
                        //---------------------------------------------//

                        //Set up timer for the next stream packet according to the requested bitrate
                        double packet_interval=(datapacket->getBitLength()/(datarate->doubleValue()*1000000));

                        //Check for huge intervals and mimize them
                        if(packet_interval>100)
                                packet_interval=100;

                        simtime_t interval=packet_interval;             // We are talking about Mbps in the datarate
                        scheduleAt(simTime()+interval, timer);

                        send(datapacket, "ifOut", 0);
                        iter->second->numPkSent++;
                        packetsSent++;
                }
                //If the is no FID delete the timer self message
                else {
                        iter->second->active=false;
                }
        }
}

void PEtherApp::setNewFID	(	InformationItem *	II,
		string	FID,
		QoSList	qos,
		vector< string >	lidpath
	)			[virtual]

A public function of this class that allows other modules to fill in new FIDs for the published Information Items of this application. It is usually called by the Topology Manager, in order to inform the publisher of the new FID for an Information Item currently being published.

                                                                                              {

        Enter_Method("TM informing %s with new FID",this->getParentModule()->getFullName());

        //Check is the publishing application supports this information item
        subscriptionItem subitem;
        subitem.ii=II;
        subscrIter subiter=find(subscriptions.begin(),subscriptions.end(),subitem);

        //If the information item was found
        if(subiter!=subscriptions.end()) {
                //Correct the QoS requirements in case they are wrong
                subiter->qos=qos;

                //Check if Information Item is already in the FID table
                FIDIter iter=FID_map.find(II);
                //If the Information Item already exists in FID Table
                if(iter!=FID_map.end()) {

                        //It doesn't matter if the FID is empty
                        iter->second->FID=FID;

                        //It doesn't matter if the LID path is empty
                        iter->second->path=lidpath;

                        //If the stream is not active right now, create new timer message right away
                        if(iter->second->active!=true) {
                                iter->second->active=true;

                                //Cancel the future scheduling for this stream
                                iter->second->tmsg=cancelEvent(iter->second->tmsg);
                                //First packet right away
                                sendStreamData(iter->second->tmsg);
                        }
                }
                //If the Information Item doesn't have an FID record, create it
                else {
                        StreamData *d = new StreamData();
                        d->FID = FID;
                        d->path=lidpath;
                        d->numPkSent = 0;
                        d->byteSize =  par("fileSize").doubleValue();
                        d->active=true;

                        InformationItem *IIkey=new InformationItem(*II);
                        FID_map.insert(std::pair<InformationItem *, StreamData *>(IIkey,d));

                        cMessage *timer = new cMessage("StreamTmr");
                        timer->setContextPointer(IIkey);

                        //Set a pointer to the timer packet so that we can find it!
                        d->tmsg=timer;

                        // first packet right away
                        sendStreamData(timer);
                }
        }
        else
                ev << "Error: II not found in subscriptions!!!" << endl;
}

---

Member Data Documentation

MACAddress PEtherApp::destMACAddress [protected]

The publication MAC address -- Preferably Broadcast.

FIDMap PEtherApp::FID_map [protected]

The map defining which data stream options describe each Information item, the Publisher is currently transmitting.

double PEtherApp::psProportion [protected]

Proportion between the publishers and subscribers.

Stream transmission parameters

---

The documentation for this class was generated from the following files:

• PEtherApp.h
• PEtherApp.cc