arzmq-sys 0.5.4

#include "normSimAgent.h"
#include <errno.h>

// This NORM simulation agent includes support for providing transport
// of a message stream from the MGEN simulation agent with restrictions.  The
// current restriction is the MGEN simulation agent

NormSimAgent::NormSimAgent(ProtoTimerMgr&         timerMgr,
                           ProtoSocket::Notifier& socketNotifier)
  : msg_sink(NULL), session_mgr(timerMgr, socketNotifier), session(NULL),
    address(NULL), port(0), ttl(3), 
    tx_rate(NormSession::DEFAULT_TRANSMIT_RATE), probe_count(1),
    cc_enable(false), ecn_enable(false), cc_mode(NORM_FIXED),
    unicast_nacks(false), silent_receiver(false),
    backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR),
    segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0),
    group_size(NormSession::DEFAULT_GSIZE_ESTIMATE),
    grtt_estimate(NormSession::DEFAULT_GRTT_ESTIMATE),
    tx_buffer_size(1024*1024), 
    tx_cache_min(NormSession::DEFAULT_TX_CACHE_MIN), 
    tx_cache_max(NormSession::DEFAULT_TX_CACHE_MAX), 
    tx_cache_size(NormSession::DEFAULT_TX_CACHE_SIZE), 
    rx_buffer_size(1024*1024), rx_cache_max(NormSession::DEFAULT_RX_CACHE_MAX),
    tx_object_size(0), tx_object_interval(0.0), 
    tx_object_size_min(0), tx_object_size_max(0), 
    tx_repeat_count(0), tx_repeat_interval(0.0),
    tx_requeue(0), tx_requeue_count(0),
    stream(NULL), auto_stream(false), 
    stream_buffer_max(0), stream_buffer_count(0), 
    stream_bytes_remain(0), watermark_pending(false),
    flow_control(false), push_mode(false), flush_mode(NormStreamObject::FLUSH_PASSIVE),
    msg_sync(false), mgen_bytes(0), mgen_pending_bytes(0),
    tracing(false), log_file_ptr(NULL), tx_loss(0.0), rx_loss(0.0)
{
    // Bind NormSessionMgr to this agent and simulation environment
    session_mgr.SetController(static_cast<NormController*>(this));
    
    interval_timer.SetListener(this, &NormSimAgent::OnIntervalTimeout);
    interval_timer.SetInterval(0.0);
    interval_timer.SetRepeat(0);
    
    memset(mgen_buffer, 0, 64);
}

NormSimAgent::~NormSimAgent()
{
    if (address) delete address;
}


const char* const NormSimAgent::cmd_list[] = 
{
    "+debug",        // debug level
    "+log",          // log file name
    "-trace",        // message tracing
    "+flog",         // set "file" send/recv log file name
    "+txloss",       // tx packet loss percent
    "+rxloss",       // rx packet loss percent
    "+address",      // session dest address
    "+ttl",          // multicast ttl
    "+rate",         // tx rate
    "+cc",           // congestion control on/off (sets NORM-CC operation, or fixed rate if off)
    "+ecn",          // ecn support on/off
    "-cce",          // sets NORM-CCE operation
    "-ccl",          // sets NORM-CCL operation
    "+probe",        // set number of cc probes per rtt
    "+flowControl",  // flow control on/off
    "+backoff",      // backoff factor 'k' (maxBackoff = k * GRTT)
    "+input",        // stream input
    "+output",       // stream output
    "+interval",     // delay between tx objects
    "+repeat",       // number of times to repeat tx object set
    "+rinterval",    // repeat interval
    "+segment",      // sender segment size
    "+block",        // sender blocking size
    "+parity",       // sender parity segments calculated per block
    "+auto",         // sender auto parity count
    "+extra",        // sender extra parity count
    "+gsize",        // group size estimate
    "+grtt",         // grtt estimate
    "+txbuffer",     // tx buffer size (bytes)
    "+txcachebounds",// <countMin:countMax:sizeMax> limits on sender tx object caching
    "+rxbuffer",     // rx buffer size (bytes)
    "+rxcachemax",   // sets rcvr "max_pending_range" for NormObjectIds
    "+start",        // open session and begin activity
    "-stop",         // cease activity and close session
    "+sendFile",     // queue a "sim" file of <size> bytes for transmission
    "+sendRandomFile",  // queue random-size file size range <sizeMin>:<sizeMax>
    "+requeue",      // requeue (i.e. retransmit) each file sent <count> times
    "+sendStream",   // send a simulated NORM stream
    "+openStream",   // open a stream object for messaging (sending)
    "+push",         // "on" means real-time push stream advancement (non-blocking)
    "+flush",        // stream flush mode
    "-doFlush",      // invoke flushing of stream
    "+ackingNode",   // <nodeId>; add <nodeId> to acking node list   
    "+unicastNacks", // receivers will unicast feedback
    "+silentReceiver", // receivers will not transmit
    NULL         
};
    
bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
{
    CmdType type = CommandType(cmd);
    ASSERT(CMD_INVALID != type);
    unsigned int len = strlen(cmd);
    if ((CMD_ARG == type) && !val)
    {
        PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(%s) missing argument\n", cmd);
        return false;        
    }
    
    if (!strncmp("debug", cmd, len))
    {
        
        int debugLevel = atoi(val);
        if ((debugLevel < 0) || (debugLevel > 12))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(segment) invalid debug level!\n");   
            return false;
        }
        SetDebugLevel(debugLevel);
    }
    else if (!strncmp("log", cmd, len))
    {
        OpenDebugLog(val);
    }
    else if (!strncmp("flog", cmd, len))
    {
        // open a file to log "file" enqueue/completion events in ".pdrc" format
        if (NULL == (log_file_ptr = fopen(val, "w")))
        {
            perror("NormSimAgent::ProcessCommand(flog) fopen() error:");
            return false;
        }
    }
    else if (!strncmp("trace", cmd, len))
    {
        tracing = true;
        if (session) session->SetTrace(true);
    }
    else if (!strncmp("txloss", cmd, len))
    {
        double txLoss = atof(val);
        if (txLoss < 0)
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(txloss) invalid txRate!\n");   
            return false;
        }
        tx_loss = txLoss;
        if (session) session->SetTxLoss(txLoss);
    }
    else if (!strncmp("rxloss", cmd, len))
    {
        double rxLoss = atof(val);
        if (rxLoss < 0)
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(rxloss) invalid txRate!\n");   
            return false;
        }
        rx_loss = rxLoss;
        if (session) session->SetRxLoss(rxLoss);
    }
    else if (!strncmp("address", cmd, len))
    {
        unsigned int len = strlen(val);
        if (address) delete address;
        if (!(address = new char[len+1]))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(address) allocation error:%s\n",
                strerror(errno)); 
            return false;
        }
        strcpy(address, val);
        char* ptr = strchr(address, '/');
        if (!ptr)
        {
            delete address;
            address = NULL;
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(address) missing port number!\n");   
            return false;
        }
        *ptr++ = '\0';
        int portNum = atoi(ptr);
        if ((portNum < 1) || (portNum > 65535))
        {
            delete address;
            address = NULL;
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(address) invalid port number!\n");   
            return false;
        }
        port = portNum;
    }
    else if (!strncmp("ttl", cmd, len))
    {
        int ttlTemp = atoi(val);
        if ((ttlTemp < 1) || (ttlTemp > 255))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(ttl) invalid value!\n");   
            return false;
        }
        ttl = ttlTemp;
    }
    else if (!strncmp("rate", cmd, len))
    {
        double txRate = atof(val);
        if (txRate < 0)
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(rate) invalid txRate!\n");   
            return false;
        }
        tx_rate = txRate;
        if (session) session->SetTxRate(txRate);
    }
    else if (!strncmp("probe", cmd, len))
    {
        unsigned int probeCount = (unsigned int)atoi(val);
        if (0 == probeCount)
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(probe) invalid probeCount!\n");   
            return false;
        }
        probe_count = probeCount;
        if (session) session->SetProbeCount(probeCount);
    }
    else if (!strncmp("cc", cmd, len))
    {
        bool ccEnable;
        if (!strcmp(val, "on"))
        {
            ccEnable = true;
        }
        else if (!strcmp(val, "off"))
        {
            ccEnable = false;
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(cc) invalid argument!\n");   
            return false;
        }    
        if (ccEnable) 
            SetCCMode(NORM_CC);
        else
            SetCCMode(NORM_FIXED);
        return true;
    }
    else if (!strncmp("ecn", cmd, len))
    {
        if (!strcmp(val, "on"))
        {
            ecn_enable = true;
        }
        else if (!strcmp(val, "off"))
        {
            ecn_enable = false;
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(ecn) invalid argument!\n");   
            return false;
        }        
        if (session) 
        {
            bool ignoreLoss = (NORM_CCE == cc_mode);
            bool tolerateLoss = (NORM_CCL == cc_mode);
            session->SetEcnSupport(ecn_enable, ignoreLoss, tolerateLoss);
        }
        return true;
    }
    else if (!strncmp("cce", cmd, len))
    {
        // This sets up NORM-CCE congestion control
        // (i.e. ECN-enabled and "ignoreLoss)
        SetCCMode(NORM_CCE);
        return true;
    }
    else if (!strncmp("ccl", cmd, len))
    {
        // This sets up NORM-CCE congestion control
        // (i.e. ECN-enabled and "ignoreLoss)
        SetCCMode(NORM_CCL);
        return true;
    }
    else if (!strncmp("backoff", cmd, len))
    {
        double backoffFactor = atof(val);
        if (backoffFactor < 0)
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(backoff) invalid txRate!\n");   
            return false;
        }
        backoff_factor = backoffFactor;
        if (session) session->SetBackoffFactor(backoffFactor);
    }
    else if (!strncmp("interval", cmd, len))
    {
        if (1 != sscanf(val, "%lf", &tx_object_interval)) 
            tx_object_interval = -1.0;
        if (tx_object_interval < 0.0)
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(interval) Invalid tx object interval: %s\n",
                     val);
            tx_object_interval = 0.0;
            return false;
        }
    }    
    else if (!strncmp("repeat", cmd, len))
    {
        tx_repeat_count = atoi(val);  
    }
    else if (!strncmp("segment", cmd, len))
    {
        int segmentSize = atoi(val);
        if ((segmentSize < 0) || (segmentSize > 8300))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(segment) invalid segment size!\n");   
            return false;
        }
        segment_size = segmentSize;
    }
    else if (!strncmp("block", cmd, len))
    {
        int blockSize = atoi(val);
        if ((blockSize < 1) || (blockSize > 255))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(block) invalid block size!\n");   
            return false;
        }
        ndata = blockSize;
    }
    else if (!strncmp("parity", cmd, len))
    {
        int numParity = atoi(val);
        if ((numParity < 0) || (numParity > 254))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(parity) invalid value!\n");   
            return false;
        }
        nparity = numParity;
    }
    else if (!strncmp("auto", cmd, len))
    {
        int autoParity = atoi(val);
        if ((autoParity < 0) || (autoParity > 254))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(auto) invalid value!\n");   
            return false;
        }
        auto_parity = autoParity;
        if (session) session->SenderSetAutoParity(autoParity);
    }
    else if (!strncmp("extra", cmd, len))
    {
        int extraParity = atoi(val);
        if ((extraParity < 0) || (extraParity > 254))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(extra) invalid value!\n");   
            return false;
        }
        extra_parity = extraParity;
        if (session) session->SenderSetExtraParity(extraParity);
    }
    else if (!strncmp("gsize", cmd, len))
    {
        if (1 != sscanf(val, "%lf", &group_size))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(gize) invalid value!\n");   
            return false;
        }
        if (session) session->SenderSetGroupSize(group_size);
    }
    else if (!strncmp("grtt", cmd, len))
    {
        if (1 != sscanf(val, "%lf", &grtt_estimate))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(gize) invalid value!\n");   
            return false;
        }
        if (session) session->SenderSetGroupSize(grtt_estimate);
    }
    else if (!strncmp("txbuffer", cmd, len))
    {
        if (1 != sscanf(val, "%lu", &tx_buffer_size))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(txbuffer) invalid value!\n");   
            return false;
        }
    }
    else if (!strncmp("txcachebounds", cmd, len))
    {
        unsigned long countMin, countMax, sizeMax;
        if (3 != sscanf(val, "%lu:%lu:%lu\n", &countMin, &countMax, &sizeMax))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(txcachebounds) invalid value!\n");   
            return false;
        }
        tx_cache_min = countMin;
        tx_cache_max = countMax;
        tx_cache_size = sizeMax;
        if (session) session->SetTxCacheBounds(tx_cache_size, tx_cache_min, tx_cache_max);
    }
    else if (!strncmp("rxbuffer", cmd, len))
    {
        if (1 != sscanf(val, "%lu", &rx_buffer_size))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(rxbuffer) invalid value!\n");   
            return false;
        }
    }
    else if (!strncmp("rxcachemax", cmd, len))
    {
        if (1 != sscanf(val, "%lu", &rx_cache_max))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(rxcachemax) invalid value!\n");   
            return false;
        }
        if (session) session->SetRxCacheMax(rx_cache_max);
    }
    else if (!strncmp("start", cmd, len))
    {
        if (!strcmp("sender", val))
        {
            return StartSender();
        }
        else if (!strcmp("receiver", val))
        {
           return StartReceiver(); 
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(start) invalid value!\n");   
            return false;    
        }           
    }
    else if (!strncmp("stop", cmd, len))
    {
        Stop();   
    }
    else if (!strncmp("sendFile", cmd, len))
    {
        if (1 != sscanf(val, "%lu", &tx_object_size))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendFile) invalid size!\n");   
            return false;
        }
        if (session)
        {
            if (tx_repeat_count >= 0) tx_repeat_count++;
            OnIntervalTimeout(interval_timer);
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendFile) no session started!\n");
            return false;
        }   
    }  
    else if (!strncmp("sendRandomFile", cmd, len))
    {
        if (2 != sscanf(val, "%lu:%lu", &tx_object_size_min, &tx_object_size_max))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendRandomFile) invalid size!\n");   
            return false;
        }
        if (tx_object_size_min > tx_object_size_max)
        {
            unsigned int tempSize = tx_object_size_min;
            tx_object_size_min = tx_object_size_max;
            tx_object_size_max = tempSize;
        }
        tx_object_size = tx_object_size_max - tx_object_size_min;
        tx_object_size = (unsigned long)(tx_object_size * UniformRand(1.0));
        tx_object_size += tx_object_size_min;
        if (session)
        {
            if (tx_repeat_count >= 0) tx_repeat_count++;
            OnIntervalTimeout(interval_timer);
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendRandomFile) no session started!\n");
            return false;
        }  
    }    
    else if (!strncmp("requeue", cmd, len))
    {
        if (1 != sscanf(val, "%d", &tx_requeue))
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(requeue) invalid count!\n");   
            return false;
        }
        tx_requeue_count = tx_requeue;
    }
    else if (!strncmp("sendStream", cmd, len))
    {
        if (session)
        {
            if (stream)
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendStream) stream already open!\n");   
                return false;
            }
            if (1 != sscanf(val, "%lu", &tx_object_size))
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendStream) invalid buffer size!\n");   
                return false;
            }
            if (!(stream = session->QueueTxStream(tx_object_size)))
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendStream) error opening stream!\n");
                return false;
            }
            stream->SetFlushMode(flush_mode);
            stream->SetPushMode(push_mode);
            auto_stream = true;
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(sendStream) session not started!\n");
            return false;
        }   
    }  
    else if (!strncmp("openStream", cmd, len))
    {
        if (session)
        {
            if (NULL != stream)
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(openStream) stream already open!\n");   
                return false;
            }
            if (1 != sscanf(val, "%lu", &tx_object_size))
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(openStream) invalid buffer size!\n");   
                return false;
            }
            if(!(tx_msg_buffer = new char[65536]))
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(openStream) error allocating tx_msg_buffer: %s\n",
                        strerror(errno));
                return false;
            } 
            if (!(stream = session->QueueTxStream(tx_object_size)))
            {
                PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(openStream) error opening stream!\n");
                return false;
            }
            stream->SetFlushMode(flush_mode);
            stream->SetPushMode(push_mode);
            auto_stream = false;  
            tx_msg_len = tx_msg_index = 0;
            if (flow_control)
            {
                // ACK-based flow control has been enabled, so initialize
                stream_buffer_max = ComputeStreamBufferSegmentCount(tx_object_size, segment_size, ndata);
                stream_buffer_max -= ndata; // a little safety margin
                stream_buffer_count = stream_bytes_remain = 0;
                watermark_pending = false;
            }
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(openStream) session not started!\n");
            return false;
        }   
    }  
    else if (!strncmp("flush", cmd, len))
    {
        int valLen = strlen(val);
        if (!strncmp("none", val, valLen))
            flush_mode = NormStreamObject::FLUSH_NONE;
        else if (!strncmp("passive", val, valLen))
            flush_mode = NormStreamObject::FLUSH_PASSIVE;
        else if (!strncmp("active", val, valLen))
            flush_mode = NormStreamObject::FLUSH_ACTIVE;
        else
        {
            PLOG(PL_FATAL, "NormApp::OnCommand(flush) invalid msg flush mode!\n");   
            return false;
        }
    }
    else if (!strncmp("doFlush", cmd, len))
    {
        if (session)
        {
            return FlushStream(true);  // TBD - provide control over EOM marking here?
        }
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(doFlush) session not started!\n");
            return false;
        }          
    }
    else if (!strncmp("push", cmd, len))
    {
        if (!strcmp(val, "on"))
            push_mode = true;
        else if (!strcmp(val, "off"))
            push_mode = false;
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(push) invalid argument!\n");   
            return false;
        }        
        return true;
    }
    else if (!strncmp("ackingNode", cmd, len))
    {
        return AddAckingNode(atoi(val));
    }
    else if (!strncmp("unicastNacks", cmd, len))
    {
        if (!strcmp(val, "on"))
            unicast_nacks = true;
        else if (!strcmp(val, "off"))
            unicast_nacks = false;
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(unicastNacks) invalid argument!\n");   
            return false;
        }        
        if (session) session->ReceiverSetUnicastNacks(unicast_nacks);
        return true;
    }
    else if (!strncmp("silentReceiver", cmd, len))
    {
        if (!strcmp(val, "on"))
            silent_receiver = true;
        else if (!strcmp(val, "off"))
            silent_receiver = false;
        else
        {
            PLOG(PL_FATAL, "NormSimAgent::ProcessCommand(silentReceiver) invalid argument!\n");   
            return false;
        }        
        if (session) session->ReceiverSetSilent(silent_receiver);
        return true;
    }
    return true;
}  // end NormSimAgent::ProcessCommand()

unsigned int NormSimAgent::ComputeStreamBufferSegmentCount(unsigned int bufferBytes, UINT16 segmentSize, UINT16 blockSize)
{
    // This same computation is performed in NormStreamObject::Open() in "normObject.cpp"
    unsigned int numBlocks = bufferBytes / (blockSize * segmentSize);
    if (numBlocks < 2) numBlocks = 2; // NORM enforces a 2-block minimum buffer size
    return (numBlocks * blockSize);
}  // end NormSimAgent::ComputeStreamBufferSegmentCount()

unsigned int NormSimAgent::WriteToStream(const char*        buffer,
                                         unsigned int       numBytes)
{
    ASSERT(NULL != stream);
    if (!flow_control)
    {
        return stream->Write(buffer, numBytes, false);
    }
    else if (stream_buffer_count < stream_buffer_max)
    {
        // This method uses stream->Write(), but limits writes by explicit ACK-based flow control status
    // 1) How many buffer bytes are available?
        unsigned int bytesAvailable = segment_size * (stream_buffer_max - stream_buffer_count);
        bytesAvailable -= stream_bytes_remain;  // unflushed segment portiomn
        if (numBytes <= bytesAvailable) 
        {
            unsigned int totalBytes = numBytes + stream_bytes_remain;
            unsigned int numSegments = totalBytes / segment_size;
            stream_bytes_remain = totalBytes % segment_size;
            stream_buffer_count += numSegments;
        }
        else
        {
            numBytes = bytesAvailable;
            stream_buffer_count = stream_buffer_max;        
        }
        // 2) Write to the stream
        unsigned int bytesWritten = stream->Write(buffer, numBytes, false);
        
        ASSERT(bytesWritten == numBytes);  // this could fail if timer-based flow control is left enabled
        // We had a "stream is closing" case here somehow?  Need to make sure we don't try
        // to write to norm stream
        
        // 3) Do we need to issue a watermark ACK request?
        if (!watermark_pending && (stream_buffer_count >= (stream_buffer_max >> 1)))
        {
            //TRACE("NormSimAgent::WriteToStream() initiating watermark ACK request (buffer count:%lu max:%lu usage:%u)...\n",
            //            stream_buffer_count, stream_buffer_max, NormStreamGetBufferUsage(tx_stream));
            session->SenderSetWatermark(stream->GetId(), 
                                        stream->FlushBlockId(),
                                        stream->FlushSegmentId(),
                                        false);
            watermark_pending = true;
        }
        return bytesWritten;
    }
    else
    {
        PLOG(PL_DETAIL, "NormSimAgent::WriteToStream() is blocked pending acknowledgment from receiver\n");
        return 0;
    }
}  // end NormSimAgent::WriteToStream()


void NormSimAgent::SetCCMode(NormCC ccMode)
{
    cc_mode = ccMode;
    if (NULL != session)
    {
        switch (ccMode)
        {
            case NORM_FIXED:
                cc_enable = false;
                session->SetCongestionControl(false);
                session->SetEcnSupport(ecn_enable, false, false);
                break;
            case NORM_CC:
                cc_enable = true;
                session->SetCongestionControl(true);
                session->SetEcnSupport(ecn_enable, false, false);
                break;
            case NORM_CCE:
                cc_enable = true;
                session->SetCongestionControl(true);
                session->SetEcnSupport(true, true, false);
                break;
            case NORM_CCL:
                cc_enable = true;
                session->SetCongestionControl(true);
                session->SetEcnSupport(ecn_enable, false, true);
                break;
        }
    }
}  // end NormSimAgent::SetCCMode()


NormSimAgent::CmdType NormSimAgent::CommandType(const char* cmd)
{
    if (!cmd) return CMD_INVALID;
    unsigned int len = strlen(cmd);
    CmdType type = CMD_INVALID;
    const char* const* nextCmd = cmd_list;
    int matchCount = 0;
    while (NULL != *nextCmd)
    {
        if (!strncmp(cmd, *nextCmd+1, len))
        {
            if (len == strlen(*nextCmd+1))
            {
                // exact match
                return (('+' == *nextCmd[0]) ? CMD_ARG : CMD_NOARG);
            }
            matchCount++;   
            if ('+' == *nextCmd[0])
                type = CMD_ARG;
            else
                type = CMD_NOARG;
        }
        nextCmd++;
    }
    switch (matchCount)
    {
        case 0:
        case 1:
            return type;
        default:
            return CMD_INVALID;  // it was ambiguous
    }
}  // end NormSimAgent::CommandType()

bool NormSimAgent::SendMessage(unsigned int len, const char* txBuffer)
{
    if (session)
    {
        if (NULL == stream)
        {
            if(NULL == (tx_msg_buffer = new char[65536]))
            {
                PLOG(PL_FATAL, "NormSimAgent::SendMessage() error allocating tx_msg_buffer: %s\n",
                        strerror(errno));
                return false;
            } 
            if (NULL == (stream = session->QueueTxStream(tx_object_size)))
            {
                PLOG(PL_FATAL, "NormSimAgent::SendMessage() error opening stream!\n");
                return false;
            }
            auto_stream = false;  
            tx_msg_len = tx_msg_index = 0;
        }
        
        if (0 == tx_msg_len)
        {
            memcpy(tx_msg_buffer, txBuffer, len);
            tx_msg_len = len;
            OnInputReady();
            return true;
        }
        else
        {
            // Message still pending, can't send yet
            PLOG(PL_DEBUG, "NormSimAgent::SendMessage() warning: input overflow!\n");
            //ASSERT(0);
            return false;
        }
    }
    else
    {
        PLOG(PL_FATAL, "NormSimAgent::SendMessage() session not started!\n");
        return false;
    }

    return true;
}  // end NormSimAgent::SendMessage()

void NormSimAgent::OnInputReady()
{
    if (auto_stream)
    {
        // sending a dummy byte stream (fill the stream buffer up)
        char buffer[NormMsg::MAX_SIZE];
        bool inputReady = true;
        while (inputReady)
        {
            unsigned int numBytes = WriteToStream(buffer, segment_size);
            inputReady = (numBytes == segment_size);
        }
    }
    else if (tx_msg_index < tx_msg_len)
    {
        unsigned int bytesWrote;
        if (flow_control)
        {
            bytesWrote = WriteToStream(tx_msg_buffer+tx_msg_index,
                                       tx_msg_len - tx_msg_index);
        }
        else
        {
            bytesWrote = WriteToStream(tx_msg_buffer+tx_msg_index,
                                       tx_msg_len - tx_msg_index);
        }
        tx_msg_index += bytesWrote;
        if (tx_msg_index == tx_msg_len)
        {
            // Mark EOM _and_ flush (using preset flush mode)
            FlushStream(true);  
            tx_msg_index = tx_msg_len = 0;
        }   
    }
}  // end NormSimAgent::OnInputReady()

// Do an _active_ flush of the stream (for use at end-of-transmission)
bool NormSimAgent::FlushStream(bool eom)
{
    if (stream && session && session->IsSender())
    {
        stream->Flush(eom);
        if (flow_control && (NormStreamObject::FLUSH_ACTIVE == flush_mode))
        {
            if (0 != stream_bytes_remain)
            {
                stream_buffer_count++;
                stream_bytes_remain = 0;
                if (!watermark_pending && (stream_buffer_count >= (stream_buffer_max >> 1)))
                {
                    //TRACE("NormSimAgent::FlushStream() initiating watermark ACK request (buffer count:%lu max:%lu usage:%u)...\n",
                    //            stream_buffer_count, stream_buffer_max, NormStreamGetBufferUsage(tx_stream));
                    session->SenderSetWatermark(stream->GetId(), 
                                                stream->FlushBlockId(),
                                                stream->FlushSegmentId(),
                                                false);
                    watermark_pending = true;
                }
            }
        }
        return true;
    }
    else
    {
        PLOG(PL_FATAL, "NormSimAgent::FlushStream() no output stream to flush\n");
        return false;
    }   
}  // end NormSimAgent::FlushStream()
    

void NormSimAgent::Notify(NormController::Event event,
                          class NormSessionMgr* sessionMgr,
                          class NormSession*    session,
                          class NormSenderNode* sender,
                          class NormObject*     object)
{
    switch (event)
    {
       case TX_QUEUE_VACANCY:
       case TX_QUEUE_EMPTY:
          // Can queue a new object or write to stream for transmission  
          if ((NULL != object) && (object == stream))
          {
              // Stream starved, get input  
              OnInputReady();
          }
          else
          {            
              // Schedule or queue next "sim object" transmission
              if (interval_timer.GetInterval() > 0.0)
                ActivateTimer(interval_timer);            
              else
                OnIntervalTimeout(interval_timer);
          } 
          break;
          
       case TX_WATERMARK_COMPLETED:
           if (flow_control)
           {
                if (NormSession::ACK_SUCCESS == session->SenderGetAckingStatus(NORM_NODE_ANY))
                {
                    // if we've been blocking normal input, unblock it and prompt for more
                    // (we could post a vacancy notification here?
                    watermark_pending = false;
                    bool wasBlocked = stream_buffer_count >= stream_buffer_max;
                    stream_buffer_count -= (stream_buffer_max >> 1);
                    if (wasBlocked) OnInputReady();
                }
                else
                {
                    // reset watermark request
                    session->SenderResetWatermark();
                }   
           }
           break;          
          
       case TX_OBJECT_SENT:
          if ((0 != tx_requeue) && (object != stream))
          {
            if (0 != tx_requeue_count)
            {
                if (session->RequeueTxObject(object))
                {
                    if (tx_requeue_count > 0) tx_requeue_count--;
                }
                else
                {
                    // unable to requeue
                    PLOG(PL_ERROR, "NormSimAgent::Notify(TX_OBJECT_SENT) warning!: requeue attempt exceeded configured tx cache bounds!\n");
                    // reset "tx_requeue_count" for next tx object
                    tx_requeue_count = tx_requeue;
                }
            }
            else
            {
                // reset "tx_requeue_count" for next tx object
                tx_requeue_count = tx_requeue;
            }
          }
          break;
      
    case RX_OBJECT_NEW:
      {
          // It's up to the app to "accept" the object
          switch (object->GetType())
          {
              case NormObject::STREAM:
                {
                    NormObjectSize size;
                    if (silent_receiver)
                      size = NormObjectSize(rx_buffer_size);
                    else
                      size = object->GetSize();
                    if (!((NormStreamObject*)object)->Accept(size.LSB()))
                    {
                        PLOG(PL_ERROR, "NormSimAgent::Notify(RX_OBJECT_NEW) stream object accept error!\n");
                    }
                    if (!stream)
                      stream = (NormStreamObject*)object;
                    else
                      PLOG(PL_ERROR, "NormSimAgent::Notify(RX_OBJECT_NEW) warning! one stream already accepted.\n");
                }
                break;   
              case NormObject::FILE:
                {
                    if (!((NormSimObject*)object)->Accept())
                    {
                        PLOG(PL_ERROR, "NormSimAgent::Notify(RX_OBJECT_NEW) sim object accept error!\n");
                    }
                    if (NULL != log_file_ptr)
                    {
                        // Log start of NEW receive object
                        unsigned long objSize = object->GetSize().LSB();
                        ProtoTime theTime;
                        theTime.GetCurrentTime();
                        double rxTime = theTime.GetValue();
                        double sec = fmod(rxTime, 60.0);
                        int min = (int)fmod(rxTime/60.0,60.0);
                        //int hr = (int)fmod(rxTime/3600.0, 24.0);
                        int hr = (int)(rxTime/3600.0);
                        fprintf(log_file_ptr, "%02d:%02d:%lf NEW size>%lu seq>%hu\n",
                                hr, min, sec, objSize, (UINT16)object->GetId());
                    }
                }
                break;
              case NormObject::DATA: 
                PLOG(PL_ERROR, "NormSimAgent::Notify() DATA objects not _yet_ supported...\n");      
                break;
              default:
                PLOG(PL_ERROR, "NormSimAgent::Notify() INVALID object type!\n");      
                ASSERT(0);
                break;    
          }   
          break;
      }
      
    case RX_OBJECT_INFO:
      switch(object->GetType())
      {
          case NormObject::FILE:
          case NormObject::DATA:
          case NormObject::STREAM:
          default:
            break;
      }  // end switch(object->GetType())
      break;
      
    case RX_OBJECT_UPDATED:
      switch (object->GetType())
      {
          case NormObject::FILE:
            // (TBD) update progress
            break;

          case NormObject::STREAM:
            {
                // Read the stream when it's updated  
                if (msg_sink)
                {      
                    bool dataReady = true;
                    while (dataReady)  
                    {       
                        if (0 == mgen_pending_bytes)
                        {
                            // Reading (at least part of) 2 byte MGEN msg len header
                            unsigned int want = 2 - mgen_bytes;
                            unsigned int got = want;
                            bool findMsgSync = msg_sync ? false : true;
                            if ((static_cast<NormStreamObject*>(object))->Read(mgen_buffer + mgen_bytes, 
                                                                               &got, findMsgSync))
                            {
                                mgen_bytes += got;
                                msg_sync = true;
                                if (got != want) dataReady = false;
                            }
                            else
                            {
                                if (msg_sync)
                                    {
                                    PLOG(PL_WARN, "NormSimAgent::Notify(1) detected stream break\n");
                                    //ASSERT(0);
                                    mgen_bytes = mgen_pending_bytes = 0;
                                    msg_sync = false;
                                    continue;
                                }
                                else
                                {
                                    break;
                                }
                            }
                            if (2 == mgen_bytes)
                            {
                                UINT16 msgSize;
                                memcpy(&msgSize, mgen_buffer, sizeof(UINT16));
                                mgen_pending_bytes = ntohs(msgSize) - 2;  
                            }
                        }
                        if (mgen_pending_bytes)
                        {
                            // Save the first part for MGEN logging
                            if (mgen_bytes < 64)
                            {
                                unsigned int want = MIN(mgen_pending_bytes, 62);
                                unsigned int got = want;
                                if ((static_cast<NormStreamObject*>(object))->Read(mgen_buffer+mgen_bytes, 
                                                                                   &got))
                                {
                                    mgen_pending_bytes -= got;
                                    mgen_bytes += got;
                                    if (got != want) dataReady = false;
                                }
                                else
                                {
                                    PLOG(PL_WARN, "NormSimAgent::Notify(2) detected stream break\n");
                                    mgen_bytes = mgen_pending_bytes = 0;
                                    msg_sync = false;
                                    continue;
                                }
                            }
                            while (dataReady && mgen_pending_bytes)
                            {
                                char buffer[256];
                                unsigned int want = MIN(256, mgen_pending_bytes); 
                                unsigned int got = want;
                                if ((static_cast<NormStreamObject*>(object))->Read(buffer, &got))
                                {
                                    mgen_pending_bytes -= got;
                                    mgen_bytes += got;
                                    if (got != want) dataReady = false;
                                }
                                else
                                {
                                    PLOG(PL_WARN, "NormSimAgent::Notify(3) detected stream break\n");
                                    mgen_bytes = mgen_pending_bytes = 0;
                                    msg_sync = false;
                                    break;
                                }
                            }
                            if (msg_sync && (0 == mgen_pending_bytes))
                            {
                                ProtoAddress srcAddr;
                                srcAddr.ResolveFromString(sender->GetAddress().GetHostString());
                                srcAddr.SetPort(sender->GetAddress().GetPort());
                                msg_sink->HandleMessage(mgen_buffer,mgen_bytes,srcAddr);
                                mgen_bytes = 0;   
                            }
                        }  // end if (mgen_pending_bytes)
                    }  // end while(dataReady)
                }
                else
                {
                    // Simply read and discard the byte stream for sim purposes
                    char buffer[1024];
                    unsigned int want = 1024;
                    unsigned int got = want;
                    while (1)
                    {
                        if ((static_cast<NormStreamObject*>(object))->Read(buffer, &got))
                        {
                            // Break when data is no longer available
                            if (got != want) break;
                        }
                        else
                        {
                            PLOG(PL_WARN, "NormSimAgent::Notify() detected stream break\n");
                        }
                        got = want = 1024;
                    }
                }
                break;
            }

          case NormObject::DATA: 
            PLOG(PL_FATAL, "NormSimAgent::Notify() DATA objects not supported...\n");      
            ASSERT(0);
            break;

          default:
            // should never occur
            break;
      }  // end switch (object->GetType())
      break;
    case RX_OBJECT_COMPLETED:
      {
          switch(object->GetType())
          {
              case NormObject::FILE:
                //DMSG(0, "norm: Completed rx file: %s\n", ((NormFileObject*)object)->Path());
                if (NULL != log_file_ptr)
                {
                    unsigned long objSize = object->GetSize().LSB();
                    ProtoTime theTime;
                    theTime.GetCurrentTime();
                    double rxTime = theTime.GetValue();
                    double sec = fmod(rxTime, 60.0);
                    int min = (int)fmod(rxTime/60.0,60.0);
                    int hr = (int)fmod(rxTime/3600.0, 24.0);
                    fprintf(log_file_ptr, "%02d:%02d:%lf RECV size>%lu seq>%hu\n",
                            hr, min, sec, objSize, (UINT16)object->GetId());
                }
                break;
              case NormObject::STREAM:
                //DMSG(0, "norm: Completed rx stream ...\n");
                break;
              case NormObject::DATA:
                ASSERT(0);
                break;
              default:
                break;
          }
          break;
      }
      case RX_OBJECT_ABORTED:
          switch(object->GetType())
          {
              case NormObject::FILE:
                //DMSG(0, "norm: Aborted rx file: %s\n", ((NormFileObject*)object)->Path());
                if (NULL != log_file_ptr)
                {
                    unsigned long objSize = object->GetSize().LSB();
                    ProtoTime theTime;
                    theTime.GetCurrentTime();
                    double rxTime = theTime.GetValue();
                    double sec = fmod(rxTime, 60.0);
                    int min = (int)fmod(rxTime/60.0,60.0);
                    int hr = (int)fmod(rxTime/3600.0, 24.0);
                    fprintf(log_file_ptr, "%02d:%02d:%lf ABORT size>%lu seq>%hu\n",
                            hr, min, sec, objSize, (UINT16)object->GetId());
                }
                break;
              case NormObject::STREAM:
                //DMSG(0, "norm: Aborted rx stream ...\n");
                break;
              case NormObject::DATA:
                ASSERT(0);
                break;
              default:
                break;
          }
          break;
          
      default:
          PLOG(PL_DEBUG, "NormSimAgent::Notify() unhandled NormEvent type\n");
        break;
    }  // end switch(event)
}  // end NormSimAgent::Notify()


bool NormSimAgent::OnIntervalTimeout(ProtoTimer& theTimer)
{
    if (tx_repeat_count)
    {
        if (stream)
        {
            // (TBD)
        }
        else
        {
            // Queue a NORM_OBJECT_SIM as long as there are repeats
            if (tx_object_size_min > 0)
            {
                tx_object_size = tx_object_size_max - tx_object_size_min;
                tx_object_size = (unsigned long)(tx_object_size * UniformRand(1.0));
                tx_object_size += tx_object_size_min;
            }
            NormObject* object = session->QueueTxSim(tx_object_size);
            if (NULL != object)
            {
                if (tx_repeat_count > 0) tx_repeat_count--;
                PLOG(PL_DEBUG, "NormSimAgent::OnIntervalTimeout(() Queued file size: %lu bytes\n", tx_object_size);  
                if (NULL != log_file_ptr)
                {
                    ProtoTime theTime;
                    theTime.GetCurrentTime();
                    double rxTime = theTime.GetValue();
                    double sec = fmod(rxTime, 60.0);
                    int min = (int)fmod(rxTime/60.0,60.0);
                    int hr = (int)fmod(rxTime/3600.0, 24.0);
                    fprintf(log_file_ptr, "%02d:%02d:%lf SEND size>%lu seq>%hu\n",
                            hr, min, sec, tx_object_size, (UINT16)object->GetId());
                } 
            }             
            else
            {
                // Note, probably was flow controlled, so we don't decrement the repeat count here and the
                // next QUEUE_VACANCY event will trigger another attempt to enqueue
                TRACE("NormSimAgent::OnIntervalTimeout() warning: unable to enqueue tx object. (flow control?)\n");
                PLOG(PL_WARN, "NormSimAgent::OnIntervalTimeout() warning: unable to enqueue tx object. (flow control?)\n");
            }   
        }
        interval_timer.SetInterval(tx_object_interval);
        return true;
    }
    else
    {
        // Done
        if (interval_timer.IsActive()) interval_timer.Deactivate();
        return false;
    }
}  // end NormSimAgent::OnIntervalTimeout()

bool NormSimAgent::AddAckingNode(NormNodeId nodeId)
{
    if (NULL == session)
    {
        PLOG(PL_FATAL, "NormSimAgent::AddAckingNode() error: sender not started!\n");
        return false;
    }
    flow_control = true;  // assume ACK flow control is desired
    return session->SenderAddAckingNode(nodeId);
}  // end NormSimAgent::AddAckingNode()
    
bool NormSimAgent::StartSender()
{
    if (NULL != session)
    {
        PLOG(PL_FATAL, "NormSimAgent::StartSender() Error! sender or receiver already started!\n");
        return false;
    }
    // Validate our session settings
    if (NULL == address)
    {
        PLOG(PL_FATAL, "NormSimAgent::StartSender() Error! no session address given.");
        return false;
    }
    
    // Create a new session on multicast group/port
    session = session_mgr.NewSession(address, port, GetAgentId());
    if (NULL != session)
    {
        // Common session parameters
        session->SetTxRate(tx_rate);
        SetCCMode(cc_mode); // calls NormSession::SetCongestionControl() and NormSession::SetEcnSupport() as needed
        session->SetProbeCount(probe_count);
        session->SetBackoffFactor(backoff_factor);
        session->SetTrace(tracing);
        session->SetTxLoss(tx_loss);
        session->SetRxLoss(rx_loss);
        session->SenderSetGroupSize(group_size);
        session->SenderSetGrtt(grtt_estimate);        
        session->SetTxCacheBounds(tx_cache_size, tx_cache_min, tx_cache_max);
        UINT16 instanceId = (rand() * 65535) / RAND_MAX;
        if (!session->StartSender(instanceId, tx_buffer_size, segment_size, ndata, nparity))
        {
            PLOG(PL_FATAL, "NormSimAgent::OnStartup() start sender error!\n");
            session_mgr.Destroy();
            return false;
        }        
        session->SenderSetAutoParity(auto_parity);
        session->SenderSetExtraParity(extra_parity);
        return true;
    }
    else
    {
        PLOG(PL_FATAL, "NormSimAgent::StartSender() new session error!\n");
        return false;
    }
}  // end NormSimAgent::StartSender()


bool NormSimAgent::StartReceiver()
{
    
    if (NULL != session)
    {
        PLOG(PL_FATAL, "NormSimAgent::StartReceiver() Error! sender or receiver already started!\n");
        return false;
    }
    // Validate our session settings
    if (NULL == address)
    {
        PLOG(PL_FATAL, "NormSimAgent::StartReceiver() Error! no session address given.");
        return false;
    }
    
    // Create a new session on multicast group/port
    session = session_mgr.NewSession(address, port, GetAgentId());
    if (session)
    {
        // Common session parameters
        session->SetTxRate(tx_rate);
        session->SetBackoffFactor(backoff_factor);
        session->SetTrace(tracing);
        session->SetTxLoss(tx_loss);
        session->SetRxLoss(rx_loss);
        SetCCMode(cc_mode); // calls NormSession::SetCongestionControl() and NormSession::SetEcnSupport() as needed
        session->ReceiverSetUnicastNacks(unicast_nacks);
        session->ReceiverSetSilent(silent_receiver);
        session->SetRxCacheMax(rx_cache_max);
        // StartReceiver(bufferSize)
        if (!session->StartReceiver(rx_buffer_size))
        {
            PLOG(PL_FATAL, "NormSimAgent::StartReceiver() start receiver error!\n");
            session_mgr.Destroy();
            return false;
        }
        return true;
    }
    else
    {
        PLOG(PL_FATAL, "NormSimAgent::StartReceiver() new session error!\n");
        return false;
    }
}  // end NormSimAgent::StartSender()

void NormSimAgent::Stop()
{
    if (session) 
    {
        if (session->IsSender()) session->StopSender();
        if (session->IsReceiver()) session->StopReceiver();
        session_mgr.DeleteSession(session);
        session = NULL;
        stream = NULL;
    }   
    if (NULL != log_file_ptr)
    {
        fclose(log_file_ptr);
        log_file_ptr = NULL;
    }
}  // end NormSimAgent::StopSender()