inet::SCTPPeer Class Reference

Implements the SCTPPeer simple module. More...

#include <SCTPPeer.h>

Inheritance diagram for inet::SCTPPeer:

Classes
struct	PathStatus

Public Member Functions
	SCTPPeer ()
	~SCTPPeer ()
Public Member Functions inherited from inet::SCTPSocket::CallbackInterface
virtual	~CallbackInterface ()
virtual void	socketEstablished (int assocId, void *yourPtr, unsigned long int buffer)
virtual void	socketDeleted (int assocId, void *yourPtr)
virtual void	sendqueueAbatedArrived (int connId, unsigned long int buffer)
virtual void	addressAddedArrived (int assocId, L3Address localAddr, L3Address remoteAddr)
Public Member Functions inherited from inet::ILifecycle
virtual	~ILifecycle ()

Protected Types
typedef std::map< int, long >	RcvdPacketsPerAssoc
typedef std::map< int, long >	SentPacketsPerAssoc
typedef std::map< int, long >	RcvdBytesPerAssoc
typedef std::map< int, cOutVector * >	BytesPerAssoc
typedef std::map< int, cDoubleHistogram * >	HistEndToEndDelay
typedef std::map< int, cOutVector * >	EndToEndDelay
typedef std::map< L3Address, PathStatus >	SCTPPathStatus

Protected Member Functions
virtual void	initialize (int stage) override
virtual int	numInitStages () const override
virtual void	handleMessage (cMessage *msg) override
virtual void	finish () override
void	handleTimer (cMessage *msg)
void	connect ()
void	socketEstablished (int connId, void *yourPtr)
void	socketDataArrived (int connId, void *yourPtr, cPacket *msg, bool urgent) override
void	socketDataNotificationArrived (int connId, void *yourPtr, cPacket *msg) override
void	socketPeerClosed (int connId, void *yourPtr) override
void	socketClosed (int connId, void *yourPtr) override
void	socketFailure (int connId, void *yourPtr, int code) override
void	socketStatusArrived (int connId, void *yourPtr, SCTPStatusInfo *status) override
void	sendRequest (bool last=true)
void	sendOrSchedule (cMessage *msg)
void	generateAndSend (SCTPConnectInfo *connectInfo)
void	sendRequestArrived () override
void	sendQueueRequest ()
void	shutdownReceivedArrived (int connId) override
void	sendqueueFullArrived (int connId) override
void	msgAbandonedArrived (int assocId) override
void	setStatusString (const char *s)
virtual bool	handleOperationStage (LifecycleOperation *operation, int stage, IDoneCallback *doneCallback) override
	Perform one stage of a lifecycle operation. More...

Protected Attributes
double	delay
bool	echo
bool	ordered
bool	schedule
int	queueSize
int	outboundStreams
int	inboundStreams
SCTPPathStatus	sctpPathStatus
SCTPSocket	clientSocket
SCTPSocket	listeningSocket
cMessage *	timeoutMsg
cMessage *	timeMsg
cMessage *	connectTimer
bool	shutdownReceived
bool	sendAllowed
int	serverAssocId
int	numRequestsToSend
int	lastStream
int	numPacketsToReceive
RcvdPacketsPerAssoc	rcvdPacketsPerAssoc
SentPacketsPerAssoc	sentPacketsPerAssoc
RcvdBytesPerAssoc	rcvdBytesPerAssoc
BytesPerAssoc	bytesPerAssoc
HistEndToEndDelay	histEndToEndDelay
EndToEndDelay	endToEndDelay
long	bytesSent
int	echoedBytesSent
int	packetsSent
int	bytesRcvd
int	packetsRcvd
int	notificationsReceived
int	numSessions
int	chunksAbandoned

Static Protected Attributes
static simsignal_t	sentPkSignal = registerSignal("sentPk")
static simsignal_t	echoedPkSignal = registerSignal("echoedPk")
static simsignal_t	rcvdPkSignal = registerSignal("rcvdPk")

Detailed Description

Implements the SCTPPeer simple module.

See the NED file for more info.

Member Typedef Documentation

typedef std::map<int, cOutVector *> inet::SCTPPeer::BytesPerAssoc

protected

typedef std::map<int, cOutVector *> inet::SCTPPeer::EndToEndDelay

protected

typedef std::map<int, cDoubleHistogram *> inet::SCTPPeer::HistEndToEndDelay

protected

typedef std::map<int, long> inet::SCTPPeer::RcvdBytesPerAssoc

protected

typedef std::map<int, long> inet::SCTPPeer::RcvdPacketsPerAssoc

protected

typedef std::map<L3Address, PathStatus> inet::SCTPPeer::SCTPPathStatus

protected

typedef std::map<int, long> inet::SCTPPeer::SentPacketsPerAssoc

protected

Constructor & Destructor Documentation

inet::SCTPPeer::SCTPPeer

(

)

{
    timeoutMsg = nullptr;
    timeMsg = nullptr;
    connectTimer = nullptr;
    delay = 0;
    echo = false;
    ordered = true;
    schedule = false;
    queueSize = 0;
    outboundStreams = 1;
    inboundStreams = 17;
    shutdownReceived = false;
    sendAllowed = true;
    serverAssocId = 0;
    numRequestsToSend = 0;
    lastStream = 0;
    numPacketsToReceive = 0;
    bytesSent = 0;
    echoedBytesSent = 0;
    packetsSent = 0;
    bytesRcvd = 0;
    packetsRcvd = 0;
    notificationsReceived = 0;
    numSessions = 0;
    chunksAbandoned = 0;
}

inet::SCTPPeer::~SCTPPeer

(

)

{
    cancelAndDelete(timeMsg);
    cancelAndDelete(timeoutMsg);
    cancelAndDelete(connectTimer);
    for (auto & elem : bytesPerAssoc)
        delete elem.second;
    bytesPerAssoc.clear();

    for (auto & elem : endToEndDelay)
        delete elem.second;
    endToEndDelay.clear();

    for (auto & elem : histEndToEndDelay)
        delete elem.second;
    histEndToEndDelay.clear();

    rcvdPacketsPerAssoc.clear();
    sentPacketsPerAssoc.clear();
    rcvdBytesPerAssoc.clear();
}

Member Function Documentation

void inet::SCTPPeer::connect ( )

protected

Referenced by handleTimer().

{
    const char *connectAddress = par("connectAddress");
    int connectPort = par("connectPort");
    int outStreams = par("outboundStreams");
    clientSocket.setOutboundStreams(outStreams);

    EV_INFO << "issuing OPEN command\n";
    EV_INFO << "Assoc " << clientSocket.getConnectionId() << "::connect to address " << connectAddress << ", port " << connectPort << "\n";
    numSessions++;
    bool streamReset = par("streamReset");
    L3Address destination;
    L3AddressResolver().tryResolve(connectAddress, destination);
    if (destination.isUnspecified())
        EV << "cannot resolve destination address: " << connectAddress << endl;
    else {
        clientSocket.connect(destination, connectPort, streamReset, (int)par("prMethod"), (unsigned int)par("numRequestsPerSession"));
    }

    if (streamReset) {
        cMessage *cmsg = new cMessage("StreamReset");
        cmsg->setKind(MSGKIND_RESET);
        EV_INFO << "StreamReset Timer scheduled at " << simTime() << "\n";
        scheduleAt(simTime() + par("streamRequestTime"), cmsg);
    }

    unsigned int streamNum = 0;
    cStringTokenizer tokenizer(par("streamPriorities").stringValue());
    while (tokenizer.hasMoreTokens()) {
        const char *token = tokenizer.nextToken();
        clientSocket.setStreamPriority(streamNum, (unsigned int)atoi(token));

        streamNum++;
    }
}

void inet::SCTPPeer::finish ( )

overrideprotectedvirtual

{
    EV_INFO << getFullPath() << ": opened " << numSessions << " sessions\n";
    EV_INFO << getFullPath() << ": sent " << bytesSent << " bytes in " << packetsSent << " packets\n";

    for (auto & elem : rcvdBytesPerAssoc)
        EV_DETAIL << getFullPath() << ": received " << elem.second << " bytes in assoc " << elem.first << "\n";

    EV_INFO << getFullPath() << "Over all " << packetsRcvd << " packets received\n ";
    EV_INFO << getFullPath() << "Over all " << notificationsReceived << " notifications received\n ";
}

void inet::SCTPPeer::generateAndSend ( SCTPConnectInfo *  connectInfo )

protected

Referenced by handleMessage(), and handleTimer().

{
    cPacket *cmsg = new cPacket("SCTP_C_SEND");
    SCTPSimpleMessage *msg = new SCTPSimpleMessage("Server");
    int numBytes = par("requestLength");
    msg->setDataArraySize(numBytes);

    for (int i = 0; i < numBytes; i++)
        msg->setData(i, 's');

    msg->setDataLen(numBytes);
    msg->setByteLength(numBytes);
    msg->setEncaps(false);
    cmsg->encapsulate(msg);
    SCTPSendInfo *cmd = new SCTPSendInfo();
    cmd->setAssocId(serverAssocId);

    if (ordered)
        cmd->setSendUnordered(COMPLETE_MESG_ORDERED);
    else
        cmd->setSendUnordered(COMPLETE_MESG_UNORDERED);

    lastStream = (lastStream + 1) % outboundStreams;
    cmd->setSid(lastStream);
    cmd->setPrValue(par("prValue"));
    cmd->setPrMethod(par("prMethod"));
    cmd->setLast(true);
    cmsg->setKind(SCTP_C_SEND);
    cmsg->setControlInfo(cmd);
    packetsSent++;
    bytesSent += msg->getByteLength();
    emit(sentPkSignal, msg);
    sendOrSchedule(cmsg);
}

void inet::SCTPPeer::handleMessage ( cMessage *  msg )

overrideprotectedvirtual

{
    int id = -1;

    if (msg->isSelfMessage())
        handleTimer(msg);

    switch (msg->getKind()) {
        case SCTP_I_PEER_CLOSED:
        case SCTP_I_ABORT: {
            SCTPCommand *ind = check_and_cast<SCTPCommand *>(msg->getControlInfo()->dup());
            cMessage *cmsg = new cMessage("SCTP_C_ABORT");
            SCTPSendInfo *cmd = new SCTPSendInfo();
            id = ind->getAssocId();
            cmd->setAssocId(id);
            cmd->setSid(ind->getSid());
            cmd->setNumMsgs(ind->getNumMsgs());
            cmsg->setControlInfo(cmd);
            delete ind;
            delete msg;
            cmsg->setKind(SCTP_C_ABORT);
            sendOrSchedule(cmsg);
            break;
        }

        case SCTP_I_ESTABLISHED: {
            if (clientSocket.getState() == SCTPSocket::CONNECTING)
                clientSocket.processMessage(PK(msg));
            else {
                SCTPConnectInfo *connectInfo = check_and_cast<SCTPConnectInfo *>(msg->removeControlInfo());
                numSessions++;
                serverAssocId = connectInfo->getAssocId();
                id = serverAssocId;
                outboundStreams = connectInfo->getOutboundStreams();
                rcvdPacketsPerAssoc[serverAssocId] = par("numPacketsToReceivePerClient");
                sentPacketsPerAssoc[serverAssocId] = par("numPacketsToSendPerClient");
                char text[50];
                sprintf(text, "App: Received Bytes of assoc %d", serverAssocId);
                bytesPerAssoc[serverAssocId] = new cOutVector(text);
                rcvdBytesPerAssoc[serverAssocId] = 0;
                sprintf(text, "App: EndToEndDelay of assoc %d", serverAssocId);
                endToEndDelay[serverAssocId] = new cOutVector(text);
                sprintf(text, "Hist: EndToEndDelay of assoc %d", serverAssocId);
                histEndToEndDelay[serverAssocId] = new cDoubleHistogram(text);

                //delete connectInfo;
                delete msg;

                if (par("numPacketsToSendPerClient").longValue() > 0) {
                    auto i = sentPacketsPerAssoc.find(serverAssocId);
                    numRequestsToSend = i->second;
                    if (par("thinkTime").doubleValue() > 0) {
                        generateAndSend(connectInfo);
                        timeoutMsg->setKind(SCTP_C_SEND);
                        scheduleAt(simTime() + par("thinkTime"), timeoutMsg);
                        numRequestsToSend--;
                        i->second = numRequestsToSend;
                    }
                    else {
                        if (queueSize == 0) {
                            while (numRequestsToSend > 0) {
                                generateAndSend(connectInfo);
                                numRequestsToSend--;
                                i->second = numRequestsToSend;
                            }
                        }
                        else if (queueSize > 0) {
                            int count = 0;
                            while (numRequestsToSend > 0 && count++ < queueSize * 2) {
                                generateAndSend(connectInfo);
                                numRequestsToSend--;
                                i->second = numRequestsToSend;
                            }

                            cMessage *cmsg = new cMessage("SCTP_C_QUEUE_MSGS_LIMIT");
                            SCTPInfo *qinfo = new SCTPInfo();
                            qinfo->setText(queueSize);
                            cmsg->setKind(SCTP_C_QUEUE_MSGS_LIMIT);
                            qinfo->setAssocId(id);
                            cmsg->setControlInfo(qinfo);
                            sendOrSchedule(cmsg);
                        }

                        EV_INFO << "!!!!!!!!!!!!!!!All data sent from Server !!!!!!!!!!\n";

                        auto j = rcvdPacketsPerAssoc.find(serverAssocId);
                        if (j->second == 0 && par("waitToClose").doubleValue() > 0) {
                            char as[5];
                            sprintf(as, "%d", serverAssocId);
                            cMessage *abortMsg = new cMessage(as);
                            abortMsg->setKind(SCTP_I_ABORT);
                            scheduleAt(simTime() + par("waitToClose"), abortMsg);
                        }
                        else {
                            EV_INFO << "no more packets to send, call shutdown for assoc " << serverAssocId << "\n";
                            cMessage *cmsg = new cMessage("SCTP_C_SHUTDOWN");
                            SCTPCommand *cmd = new SCTPCommand();
                            cmsg->setKind(SCTP_C_SHUTDOWN);
                            cmd->setAssocId(serverAssocId);
                            cmsg->setControlInfo(cmd);
                            sendOrSchedule(cmsg);
                        }
                    }
                }
            }
            break;
        }

        case SCTP_I_DATA_NOTIFICATION: {
            notificationsReceived++;
            SCTPCommand *ind = check_and_cast<SCTPCommand *>(msg->removeControlInfo());
            cMessage *cmsg = new cMessage("SCTP_C_RECEIVE");
            SCTPSendInfo *cmd = new SCTPSendInfo();
            id = ind->getAssocId();
            cmd->setAssocId(id);
            cmd->setSid(ind->getSid());
            cmd->setNumMsgs(ind->getNumMsgs());
            cmsg->setKind(SCTP_C_RECEIVE);
            cmsg->setControlInfo(cmd);
            delete ind;
            delete msg;
            if (!cmsg->isScheduled() && schedule == false) {
                scheduleAt(simTime() + par("delayFirstRead"), cmsg);
            }
            else if (schedule == true)
                sendOrSchedule(cmsg);
            break;
        }

        case SCTP_I_DATA: {
            SCTPCommand *ind = check_and_cast<SCTPCommand *>(msg->getControlInfo());
            id = ind->getAssocId();
            auto j = rcvdBytesPerAssoc.find(id);
            if (j == rcvdBytesPerAssoc.end() && (clientSocket.getState() == SCTPSocket::CONNECTED))
                clientSocket.processMessage(PK(msg));
            else if (j != rcvdBytesPerAssoc.end()) {
                j->second += PK(msg)->getByteLength();
                auto k = bytesPerAssoc.find(id);
                k->second->record(j->second);
                packetsRcvd++;

                if (!echo) {
                    if (par("numPacketsToReceivePerClient").longValue() > 0) {
                        auto i = rcvdPacketsPerAssoc.find(id);
                        i->second--;
                        SCTPSimpleMessage *smsg = check_and_cast<SCTPSimpleMessage *>(msg);
                        auto j = endToEndDelay.find(id);
                        j->second->record(simTime() - smsg->getCreationTime());
                        auto k = histEndToEndDelay.find(id);
                        k->second->collect(simTime() - smsg->getCreationTime());

                        if (i->second == 0) {
                            cMessage *cmsg = new cMessage("SCTP_C_NO_OUTSTANDING");
                            SCTPInfo *qinfo = new SCTPInfo();
                            cmsg->setKind(SCTP_C_NO_OUTSTANDING);
                            qinfo->setAssocId(id);
                            cmsg->setControlInfo(qinfo);
                            sendOrSchedule(cmsg);
                        }
                    }
                    delete msg;
                }
                else {
                    SCTPSendInfo *cmd = new SCTPSendInfo();
                    cmd->setAssocId(id);

                    SCTPSimpleMessage *smsg = check_and_cast<SCTPSimpleMessage *>(msg);
                    auto j = endToEndDelay.find(id);
                    j->second->record(simTime() - smsg->getCreationTime());
                    auto k = histEndToEndDelay.find(id);
                    k->second->collect(simTime() - smsg->getCreationTime());
                    cPacket *cmsg = new cPacket("SCTP_C_SEND");
                    bytesSent += smsg->getByteLength();
                    cmd->setPrValue(0);
                    emit(sentPkSignal, smsg);
                    cmd->setSendUnordered(cmd->getSendUnordered());
                    lastStream = (lastStream + 1) % outboundStreams;
                    cmd->setSid(lastStream);
                    cmd->setLast(true);
                    cmsg->encapsulate(smsg);
                    cmsg->setKind(SCTP_C_SEND);
                    cmsg->setControlInfo(cmd);
                    packetsSent++;
                    sendOrSchedule(cmsg);
                }
            } else {
                delete msg;
            }
            break;
        }

        case SCTP_I_SHUTDOWN_RECEIVED: {
            SCTPCommand *command = check_and_cast<SCTPCommand *>(msg->removeControlInfo());
            id = command->getAssocId();
            EV_INFO << "server: SCTP_I_SHUTDOWN_RECEIVED for assoc " << id << "\n";
            auto i = rcvdPacketsPerAssoc.find(id);

            if (i == rcvdPacketsPerAssoc.end() && (clientSocket.getState() == SCTPSocket::CONNECTED))
                clientSocket.processMessage(PK(msg));
            else if (i != rcvdPacketsPerAssoc.end()) {
                if (i->second == 0) {
                    cMessage *cmsg = new cMessage("SCTP_C_NO_OUTSTANDING");
                    SCTPInfo *qinfo = new SCTPInfo();
                    cmsg->setKind(SCTP_C_NO_OUTSTANDING);
                    qinfo->setAssocId(id);
                    cmsg->setControlInfo(qinfo);
                    sendOrSchedule(cmsg);
                }

                delete command;
                shutdownReceived = true;
            }
            delete msg;
            break;
        }

        case SCTP_I_SEND_STREAMS_RESETTED:
        case SCTP_I_RCV_STREAMS_RESETTED: {
            EV_INFO << "Streams have been resetted\n";
            break;
        }

        case SCTP_I_CLOSED:
            delete msg;
            break;
    }

    if (hasGUI()) {
        char buf[32];
        auto l = rcvdBytesPerAssoc.find(id);
        sprintf(buf, "rcvd: %ld bytes\nsent: %ld bytes", l->second, bytesSent);
        getDisplayString().setTagArg("t", 0, buf);
    }
}

virtual bool inet::SCTPPeer::handleOperationStage ( LifecycleOperation *  operation, int  stage, IDoneCallback *  doneCallback  )

inlineoverrideprotectedvirtual

Perform one stage of a lifecycle operation.

Processing may be done entirely within this method, or may be a longer process that involves nonzero simulation time or several events, and is triggered by this method call.

Return value: true = "done"; false = "not yet done, will invoke doneCallback when done"

Implements inet::ILifecycle.

    { Enter_Method_Silent(); throw cRuntimeError("Unsupported lifecycle operation '%s'", operation->getClassName()); return true; }

void inet::SCTPPeer::handleTimer ( cMessage *  msg )

protected

Referenced by handleMessage().

{
    EV_TRACE << "SCTPPeer::handleTimer\n";

    SCTPConnectInfo *connectInfo = dynamic_cast<SCTPConnectInfo *>(msg->getControlInfo());

    switch (msg->getKind()) {
        case MSGKIND_CONNECT:
            EV_INFO << "starting session call connect\n";
            connect();
            break;

        case SCTP_C_SEND:
            if (numRequestsToSend > 0) {
                generateAndSend(connectInfo);
                if (par("thinkTime").doubleValue() > 0)
                    scheduleAt(simTime() + par("thinkTime"), timeoutMsg);
                numRequestsToSend--;
            }
            break;

        case SCTP_I_ABORT: {
            cMessage *cmsg = new cMessage("CLOSE", SCTP_C_CLOSE);
            SCTPCommand *cmd = new SCTPCommand();
            int id = atoi(msg->getName());
            cmd->setAssocId(id);
            cmsg->setControlInfo(cmd);
            sendOrSchedule(cmsg);
        }
        break;

        case SCTP_C_RECEIVE:
            schedule = true;
            sendOrSchedule(PK(msg));
            break;

        default:
            break;
    }
}

void inet::SCTPPeer::initialize ( int  stage )

overrideprotectedvirtual

{
    cSimpleModule::initialize(stage);

    if (stage == INITSTAGE_LOCAL) {
        WATCH(numSessions);
        WATCH(packetsSent);
        WATCH(packetsRcvd);
        WATCH(bytesSent);
        WATCH(numRequestsToSend);
    }
    else if (stage == INITSTAGE_APPLICATION_LAYER) {
        // parameters
        const char *addressesString = par("localAddress");
        AddressVector addresses = L3AddressResolver().resolve(cStringTokenizer(addressesString).asVector());
        int port = par("localPort");
        echo = par("echo");
        delay = par("echoDelay");
        outboundStreams = par("outboundStreams");
        inboundStreams = par("inboundStreams");
        ordered = par("ordered").boolValue();
        queueSize = par("queueSize");
        timeoutMsg = new cMessage("SrvAppTimer");
        listeningSocket.setOutputGate(gate("sctpOut"));
        listeningSocket.setOutboundStreams(outboundStreams);
        listeningSocket.setInboundStreams(inboundStreams);

        if (addresses.size() == 0) {
            listeningSocket.bind(port);
            clientSocket.bind(port);
        }
        else {
            listeningSocket.bindx(addresses, port);
            clientSocket.bindx(addresses, port);
        }
        listeningSocket.listen(true, par("streamReset").boolValue(), par("numPacketsToSendPerClient").longValue());
        EV_DEBUG << "SCTPPeer::initialized listen port=" << port << "\n";
        clientSocket.setCallbackObject(this);
        clientSocket.setOutputGate(gate("sctpOut"));

        if ((simtime_t)par("startTime") > SIMTIME_ZERO) {    //FIXME is invalid the startTime == 0 ????
            connectTimer = new cMessage("ConnectTimer");
            connectTimer->setKind(MSGKIND_CONNECT);
            scheduleAt(par("startTime"), connectTimer);
        }

        NodeStatus *nodeStatus = dynamic_cast<NodeStatus *>(findContainingNode(this)->getSubmodule("status"));
        bool isOperational = (!nodeStatus) || nodeStatus->getState() == NodeStatus::UP;
        if (!isOperational)
            throw cRuntimeError("This module doesn't support starting in node DOWN state");
    }
}

void inet::SCTPPeer::msgAbandonedArrived ( int  assocId )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    chunksAbandoned++;
}

virtual int inet::SCTPPeer::numInitStages ( ) const

inlineoverrideprotectedvirtual

{ return NUM_INIT_STAGES; }

void inet::SCTPPeer::sendOrSchedule ( cMessage *  msg )

protected

Referenced by generateAndSend(), handleMessage(), and handleTimer().

{
    if (delay == 0) {
        send(msg, "sctpOut");
    }
    else {
        scheduleAt(simTime() + delay, msg);
    }
}

void inet::SCTPPeer::sendqueueFullArrived ( int  connId )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    sendAllowed = false;
}

void inet::SCTPPeer::sendQueueRequest ( )

protected

Referenced by socketEstablished().

{
    cMessage *cmsg = new cMessage("SCTP_C_QUEUE_MSGS_LIMIT");
    SCTPInfo *qinfo = new SCTPInfo();
    qinfo->setText(queueSize);
    cmsg->setKind(SCTP_C_QUEUE_MSGS_LIMIT);
    qinfo->setAssocId(clientSocket.getConnectionId());
    cmsg->setControlInfo(qinfo);
    clientSocket.sendRequest(cmsg);
}

void inet::SCTPPeer::sendRequest ( bool  last = true )

protected

Referenced by sendRequestArrived(), and socketEstablished().

{
    EV_INFO << "sending request, " << numRequestsToSend - 1 << " more to go\n";
    long numBytes = par("requestLength");

    if (numBytes < 1)
        numBytes = 1;

    EV_INFO << "SCTPClient: sending " << numBytes << " data bytes\n";

    cPacket *cmsg = new cPacket("SCTP_C_SEND");
    SCTPSimpleMessage *msg = new SCTPSimpleMessage("data");

    msg->setDataArraySize(numBytes);

    for (int i = 0; i < numBytes; i++)
        msg->setData(i, 'a');

    msg->setDataLen(numBytes);
    msg->setEncaps(false);
    msg->setBitLength(numBytes * 8);
    msg->setCreationTime(simTime());
    cmsg->encapsulate(msg);
    cmsg->setKind(ordered ? SCTP_C_SEND_ORDERED : SCTP_C_SEND_UNORDERED);

    SCTPSendInfo* sendCommand = new SCTPSendInfo;
    sendCommand->setLast(last);
    cmsg->setControlInfo(sendCommand);
    
    // send SCTPMessage with SCTPSimpleMessage enclosed
    emit(sentPkSignal, msg);
    clientSocket.sendMsg(cmsg);
    bytesSent += numBytes;
}

void inet::SCTPPeer::sendRequestArrived ( )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    int count = 0;

    EV_INFO << "sendRequestArrived numRequestsToSend=" << numRequestsToSend << "\n";

    while (numRequestsToSend > 0 && count++ < queueSize && sendAllowed) {
        numRequestsToSend--;
        sendRequest(count == queueSize || numRequestsToSend == 0);
        if (numRequestsToSend == 0) {
            EV_INFO << "no more packets to send, call shutdown\n";
            clientSocket.shutdown();
        }
    }
}

void inet::SCTPPeer::setStatusString ( const char *  s )

protected

Referenced by socketClosed(), socketDataArrived(), socketEstablished(), socketFailure(), and socketPeerClosed().

{
    if (hasGUI())
        getDisplayString().setTagArg("t", 0, s);
}

void inet::SCTPPeer::shutdownReceivedArrived ( int  connId )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    if (numRequestsToSend == 0) {
        cMessage *cmsg = new cMessage("SCTP_C_NO_OUTSTANDING");
        SCTPInfo *qinfo = new SCTPInfo();
        cmsg->setKind(SCTP_C_NO_OUTSTANDING);
        qinfo->setAssocId(connId);
        cmsg->setControlInfo(qinfo);
        clientSocket.sendNotification(cmsg);
    }
}

void inet::SCTPPeer::socketClosed ( int  connId, void *  yourPtr  )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    // *redefine* to start another session etc.
    EV_INFO << "connection closed\n";
    setStatusString("closed");
}

void inet::SCTPPeer::socketDataArrived ( int  connId, void *  yourPtr, cPacket *  msg, bool  urgent  )

overrideprotectedvirtual

Implements inet::SCTPSocket::CallbackInterface.

{
    // *redefine* to perform or schedule next sending
    packetsRcvd++;

    EV_INFO << "Client received packet Nr " << packetsRcvd << " from SCTP\n";

    SCTPCommand *ind = check_and_cast<SCTPCommand *>(msg->getControlInfo());

    emit(rcvdPkSignal, msg);
    bytesRcvd += msg->getByteLength();

    if (echo) {
        SCTPSimpleMessage *smsg = check_and_cast<SCTPSimpleMessage *>(msg);
        cPacket *cmsg = new cPacket("SCTP_C_SEND");
        echoedBytesSent += smsg->getByteLength();
        emit(echoedPkSignal, smsg);
        cmsg->encapsulate(smsg);
        cmsg->setKind(ind->getSendUnordered() ? SCTP_C_SEND_UNORDERED : SCTP_C_SEND_ORDERED);
        packetsSent++;
        clientSocket.sendMsg(cmsg);
    }

    if (par("numPacketsToReceive").longValue() > 0) {
        numPacketsToReceive--;
        if (numPacketsToReceive == 0) {
            setStatusString("closing");
            clientSocket.close();
        }
    }
}

void inet::SCTPPeer::socketDataNotificationArrived ( int  connId, void *  yourPtr, cPacket *  msg  )

overrideprotectedvirtual

Implements inet::SCTPSocket::CallbackInterface.

{
    SCTPCommand *ind = check_and_cast<SCTPCommand *>(msg->removeControlInfo());
    cMessage *cmsg = new cMessage("CMSG");
    SCTPSendInfo *cmd = new SCTPSendInfo();
    cmd->setAssocId(ind->getAssocId());
    cmd->setSid(ind->getSid());
    cmd->setNumMsgs(ind->getNumMsgs());
    cmsg->setKind(SCTP_C_RECEIVE);
    cmsg->setControlInfo(cmd);
    delete ind;
    clientSocket.sendNotification(cmsg);
}

void inet::SCTPPeer::socketEstablished ( int  connId, void *  yourPtr  )

protected

{
    int count = 0;
    // *redefine* to perform or schedule first sending
    EV_INFO << "SCTPClient: connected\n";
    setStatusString("connected");
    // determine number of requests in this session
    numRequestsToSend = par("numRequestsPerSession");
    numPacketsToReceive = par("numPacketsToReceive");

    if (numRequestsToSend < 1)
        numRequestsToSend = 0;

    // perform first request (next one will be sent when reply arrives)
    if (numRequestsToSend > 0) {
        if (par("thinkTime").doubleValue() > 0) {
            if (sendAllowed) {
                sendRequest();
                numRequestsToSend--;
            }

            timeMsg->setKind(MSGKIND_SEND);
            scheduleAt(simTime() + par("thinkTime"), timeMsg);
        }
        else {
            if (queueSize > 0) {
                while (numRequestsToSend > 0 && count++ < queueSize * 2 && sendAllowed) {
                    sendRequest(count == queueSize * 2);
                    numRequestsToSend--;
                }

                if (numRequestsToSend > 0 && sendAllowed)
                    sendQueueRequest();
            }
            else {
                while (numRequestsToSend > 0 && sendAllowed) {
                    sendRequest();
                    numRequestsToSend--;
                }
            }

            if (numPacketsToReceive == 0 && par("waitToClose").doubleValue() > 0) {
                timeMsg->setKind(MSGKIND_ABORT);
                scheduleAt(simTime() + par("waitToClose"), timeMsg);
            }

            if (numRequestsToSend == 0 && par("waitToClose").doubleValue() == 0) {
                EV_INFO << "socketEstablished:no more packets to send, call shutdown\n";
                clientSocket.shutdown();
            }
        }
    }
}

void inet::SCTPPeer::socketFailure ( int  connId, void *  yourPtr, int  code  )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    // subclasses may override this function, and add code try to reconnect after a delay.
    EV_WARN << "connection broken\n";
    setStatusString("broken");
    // reconnect after a delay
    timeMsg->setKind(MSGKIND_CONNECT);
    scheduleAt(simTime() + par("reconnectInterval"), timeMsg);
}

void inet::SCTPPeer::socketPeerClosed ( int  connId, void *  yourPtr  )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    // close the connection (if not already closed)
    if (clientSocket.getState() == SCTPSocket::PEER_CLOSED) {
        EV_INFO << "remote SCTP closed, closing here as well\n";
        setStatusString("closing");
        clientSocket.close();
    }
}

void inet::SCTPPeer::socketStatusArrived ( int  connId, void *  yourPtr, SCTPStatusInfo *  status  )

overrideprotectedvirtual

Reimplemented from inet::SCTPSocket::CallbackInterface.

{
    struct PathStatus ps;
    auto i = sctpPathStatus.find(status->getPathId());

    if (i != sctpPathStatus.end()) {
        ps = i->second;
        ps.active = status->getActive();
    }
    else {
        ps.active = status->getActive();
        ps.primaryPath = false;
        sctpPathStatus[ps.pid] = ps;
    }
    delete status;
}

Member Data Documentation

BytesPerAssoc inet::SCTPPeer::bytesPerAssoc

protected

Referenced by handleMessage(), and ~SCTPPeer().

int inet::SCTPPeer::bytesRcvd

protected

Referenced by SCTPPeer(), and socketDataArrived().

long inet::SCTPPeer::bytesSent

protected

Referenced by finish(), generateAndSend(), handleMessage(), initialize(), SCTPPeer(), and sendRequest().

int inet::SCTPPeer::chunksAbandoned

protected

Referenced by msgAbandonedArrived(), and SCTPPeer().

SCTPSocket inet::SCTPPeer::clientSocket

protected

Referenced by connect(), handleMessage(), initialize(), sendQueueRequest(), sendRequest(), sendRequestArrived(), shutdownReceivedArrived(), socketDataArrived(), socketDataNotificationArrived(), socketEstablished(), and socketPeerClosed().

cMessage* inet::SCTPPeer::connectTimer

protected

Referenced by initialize(), SCTPPeer(), and ~SCTPPeer().

double inet::SCTPPeer::delay

protected

Referenced by initialize(), SCTPPeer(), and sendOrSchedule().

bool inet::SCTPPeer::echo

protected

Referenced by handleMessage(), initialize(), SCTPPeer(), and socketDataArrived().

int inet::SCTPPeer::echoedBytesSent

protected

Referenced by SCTPPeer(), and socketDataArrived().

simsignal_t inet::SCTPPeer::echoedPkSignal = registerSignal("echoedPk")

staticprotected

Referenced by socketDataArrived().

EndToEndDelay inet::SCTPPeer::endToEndDelay

protected

Referenced by handleMessage(), and ~SCTPPeer().

HistEndToEndDelay inet::SCTPPeer::histEndToEndDelay

protected

Referenced by handleMessage(), and ~SCTPPeer().

int inet::SCTPPeer::inboundStreams

protected

Referenced by initialize(), and SCTPPeer().

int inet::SCTPPeer::lastStream

protected

Referenced by generateAndSend(), handleMessage(), and SCTPPeer().

SCTPSocket inet::SCTPPeer::listeningSocket

protected

Referenced by initialize().

int inet::SCTPPeer::notificationsReceived

protected

Referenced by finish(), handleMessage(), and SCTPPeer().

int inet::SCTPPeer::numPacketsToReceive

protected

Referenced by SCTPPeer(), socketDataArrived(), and socketEstablished().

int inet::SCTPPeer::numRequestsToSend

protected

Referenced by handleMessage(), handleTimer(), initialize(), SCTPPeer(), sendRequest(), sendRequestArrived(), shutdownReceivedArrived(), and socketEstablished().

int inet::SCTPPeer::numSessions

protected

Referenced by connect(), finish(), handleMessage(), initialize(), and SCTPPeer().

bool inet::SCTPPeer::ordered

protected

Referenced by generateAndSend(), initialize(), SCTPPeer(), and sendRequest().

int inet::SCTPPeer::outboundStreams

protected

Referenced by generateAndSend(), handleMessage(), initialize(), and SCTPPeer().

int inet::SCTPPeer::packetsRcvd

protected

Referenced by finish(), handleMessage(), initialize(), SCTPPeer(), and socketDataArrived().

int inet::SCTPPeer::packetsSent

protected

Referenced by finish(), generateAndSend(), handleMessage(), initialize(), SCTPPeer(), and socketDataArrived().

int inet::SCTPPeer::queueSize

protected

Referenced by handleMessage(), initialize(), SCTPPeer(), sendQueueRequest(), sendRequestArrived(), and socketEstablished().

RcvdBytesPerAssoc inet::SCTPPeer::rcvdBytesPerAssoc

protected

Referenced by finish(), handleMessage(), and ~SCTPPeer().

RcvdPacketsPerAssoc inet::SCTPPeer::rcvdPacketsPerAssoc

protected

Referenced by handleMessage(), and ~SCTPPeer().

simsignal_t inet::SCTPPeer::rcvdPkSignal = registerSignal("rcvdPk")

staticprotected

Referenced by socketDataArrived().

bool inet::SCTPPeer::schedule

protected

Referenced by handleMessage(), handleTimer(), and SCTPPeer().

SCTPPathStatus inet::SCTPPeer::sctpPathStatus

protected

Referenced by socketStatusArrived().

bool inet::SCTPPeer::sendAllowed

protected

Referenced by SCTPPeer(), sendqueueFullArrived(), sendRequestArrived(), and socketEstablished().

SentPacketsPerAssoc inet::SCTPPeer::sentPacketsPerAssoc

protected

Referenced by handleMessage(), and ~SCTPPeer().

simsignal_t inet::SCTPPeer::sentPkSignal = registerSignal("sentPk")

staticprotected

Referenced by generateAndSend(), handleMessage(), and sendRequest().

int inet::SCTPPeer::serverAssocId

protected

Referenced by generateAndSend(), handleMessage(), and SCTPPeer().

bool inet::SCTPPeer::shutdownReceived

protected

Referenced by handleMessage(), and SCTPPeer().

cMessage* inet::SCTPPeer::timeMsg

protected

Referenced by SCTPPeer(), socketEstablished(), socketFailure(), and ~SCTPPeer().

cMessage* inet::SCTPPeer::timeoutMsg

protected

Referenced by handleMessage(), handleTimer(), initialize(), SCTPPeer(), and ~SCTPPeer().

---

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

• SCTPPeer.h
• SCTPPeer.cc