From 1f0f97f5b54b083df9b4d6719c6fde6f1ad033e9 Mon Sep 17 00:00:00 2001 From: bebopagogo Date: Fri, 27 Dec 2019 11:40:11 -0500 Subject: [PATCH] initial addition of normCast example --- examples/normCast.cpp | 1024 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1024 insertions(+) create mode 100644 examples/normCast.cpp diff --git a/examples/normCast.cpp b/examples/normCast.cpp new file mode 100644 index 0000000..3a9a13d --- /dev/null +++ b/examples/normCast.cpp @@ -0,0 +1,1024 @@ + +#include "normApi.h" +#include // for printf(), etc + +#ifdef WIN32 +#include +//#include "win32InputHandler.cpp" // brings in the entire implementation +#else // UNIX +#include +#endif // if/else WIN32/UNIX + +#define SHOOT_NOW 1 + +// I usually avoid using Protlib stuff for NORM API examples to keep things clearer, +// but the couple of classes here are useful helpers from the Protolib C++ toolkit. + +#include "protoFile.h" // for ProtoFile::PathList and iterator for tx file/directory queue +#include "protoString.h" // for ProtoTokenator + +class NormCaster +{ + public: + NormCaster(); + ~NormCaster(); + + // some day build these directly into NORM API + enum CCMode {NORM_FIXED, NORM_CC, NORM_CCE, NORM_CCL}; + + void Destroy(); + + bool OpenNormSession(NormInstanceHandle instance, + const char* addr, + unsigned short port, + NormNodeId nodeId); + void CloseNormSession(); + + void SetNormCongestionControl(CCMode ccMode); + + void SetNormTxRate(double bitsPerSecond) + { + assert(NORM_SESSION_INVALID != norm_session); + NormSetTxRate(norm_session, bitsPerSecond); + } + void SetNormMulticastInterface(const char* ifaceName) + { + assert(NORM_SESSION_INVALID != norm_session); + NormSetMulticastInterface(norm_session, ifaceName); + } + void SetLoopback(bool state) + { + loopback = state; + if (NORM_SESSION_INVALID != norm_session) + NormSetMulticastLoopback(norm_session, state); + } + void SetNormMessageTrace(bool state) + { + assert(NORM_SESSION_INVALID != norm_session); + NormSetMessageTrace(norm_session, state); + } + void AddAckingNode(NormNodeId ackId) + { + assert(NORM_SESSION_INVALID != norm_session); + NormAddAckingNode(norm_session, ackId); + norm_acking = true; // invoke ack-based flow control + } + void SetFlushing(bool state) + {norm_flushing = state;} + + bool Start(bool sender, bool receiver); + void Stop() + {is_running = false;} + bool IsRunning() const + {return is_running;} + + void HandleNormEvent(const NormEvent& event); + + // Sender methods + bool AddTxItem(const char* path); + bool StageNextTxFile(); + bool TxFilePending() + {return '\0' != tx_pending_path[0];} + bool TxReady() const; + void SendFiles(); + bool EnqueueFileObject(); + unsigned long GetSentCount() + {return sent_count;} + + + // Receiver methods + void SetRxCacheDirectory(const char* path) + {strncpy(rx_cache_path, path, PATH_MAX);} + const char* GetRxCacheDirectory() const + {return rx_cache_path;} + + // These can only be called post-OpenNormSession() + void SetSilentReceiver(bool state) + {NormSetSilentReceiver(norm_session, true);} + void SetTxLoss(double txloss) + {NormSetTxLoss(norm_session, txloss);} + + private: + bool is_running; + NormSessionHandle norm_session; + ProtoFile::PathList tx_file_list; + ProtoFile::PathList::PathIterator tx_file_iterator; + char tx_pending_path[PATH_MAX + 1]; + unsigned int tx_pending_prefix_len; + bool is_multicast; + bool loopback; + unsigned int norm_tx_segment_size; + unsigned int norm_tx_queue_max; // max number of objects that can be enqueued at once + unsigned int norm_tx_queue_count; // count of unacknowledged enqueued objects (TBD - optionally track size too) + bool norm_flow_control_pending; + bool norm_tx_vacancy; + bool norm_acking; + bool norm_flushing; + NormObjectHandle norm_flush_object; + NormObjectHandle norm_last_object; + unsigned long sent_count; + + // receiver state variables + char rx_cache_path[PATH_MAX + 1]; + // These are some options mainly for testing purposes + //bool rx_silent; // TBD - support optional "silent receiver' modes + //double tx_loss; // cheesy, built-in way to test loss performance + +}; // end class NormCaster + +NormCaster::NormCaster() + : norm_session(NORM_SESSION_INVALID), tx_file_iterator(tx_file_list), + tx_pending_prefix_len(0), + is_multicast(false), loopback(false), norm_tx_segment_size(1400), + norm_tx_queue_max(8), norm_tx_queue_count(0), + norm_flow_control_pending(false), norm_tx_vacancy(true), norm_acking(false), + norm_flushing(true), norm_flush_object(NORM_OBJECT_INVALID), norm_last_object(NORM_OBJECT_INVALID), + sent_count(0)//, rx_silent(false), tx_loss(0.0) +{ + tx_pending_path[0] = '\0'; + tx_pending_path[PATH_MAX] = '\0'; + rx_cache_path[0] = '\0'; + rx_cache_path[PATH_MAX] = '\0'; +} + +NormCaster::~NormCaster() +{ + Destroy(); +} + +void NormCaster::Destroy() +{ + tx_file_list.Destroy(); +} + +bool NormCaster::AddTxItem(const char* path) +{ + bool result = tx_file_list.AppendPath(path); + if (!result) + perror("NormCaster::AddTxItem() error"); + if (!TxFilePending()) StageNextTxFile(); + return result; +} // end NormCaster::AddTxItem() + + +bool NormCaster::OpenNormSession(NormInstanceHandle instance, const char* addr, unsigned short port, NormNodeId nodeId) +{ + if (NormIsUnicastAddress(addr)) + is_multicast = false; + else + is_multicast = true; + norm_session = NormCreateSession(instance, addr, port, nodeId); + if (NORM_SESSION_INVALID == norm_session) + { + fprintf(stderr, "normCast error: unable to create NORM session\n"); + return false; + } + + if (is_multicast) + { + NormSetRxPortReuse(norm_session, true); + if (loopback) + NormSetMulticastLoopback(norm_session, true); + } + + // Set some default parameters (maybe we should put all parameter setting in Start()) + fprintf(stderr, "setting rx cache limit to %u\n", norm_tx_queue_max); + if (norm_tx_queue_max > 65535/2) norm_tx_queue_max = 65535/2; + NormSetRxCacheLimit(norm_session, norm_tx_queue_max); + NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL); + + if (!is_multicast) + NormSetDefaultUnicastNack(norm_session, true); + NormSetTxCacheBounds(norm_session, 10*1024*1024, norm_tx_queue_max, norm_tx_queue_max); + + //NormSetMessageTrace(norm_session, true); + + //NormSetTxRobustFactor(norm_session, 20); + + return true; +} // end NormCaster::OpenNormSession() + +void NormCaster::CloseNormSession() +{ + if (NORM_SESSION_INVALID == norm_session) return; + NormDestroySession(norm_session); + norm_session = NORM_SESSION_INVALID; +} // end NormCaster::CloseNormSession() + +void NormCaster::SetNormCongestionControl(CCMode ccMode) +{ + assert(NORM_SESSION_INVALID != norm_session); + switch (ccMode) + { + case NORM_CC: // default TCP-friendly congestion control + NormSetEcnSupport(norm_session, false, false, false); + break; + case NORM_CCE: // "wireless-ready" ECN-only congestion control + NormSetEcnSupport(norm_session, true, true); + break; + case NORM_CCL: // "loss tolerant", non-ECN congestion control + NormSetEcnSupport(norm_session, false, false, true); + break; + case NORM_FIXED: + NormSetEcnSupport(norm_session, false, false, false); + break; + } + if (NORM_FIXED != ccMode) + NormSetCongestionControl(norm_session, true); + else + NormSetCongestionControl(norm_session, false); +} // end NormCaster::SetNormCongestionControl() + +bool NormCaster::Start(bool sender, bool receiver) +{ + fprintf(stderr, "enter NormCaster::Start() ...\n"); + // TBD - make these command-line accessible + unsigned int bufferSize = 64*1024*1024; + unsigned int segmentSize = 1400; + unsigned int blockSize = 64; + unsigned int numParity = 0; + unsigned int txSockBufferSize = 4*1024*1024; + unsigned int rxSockBufferSize = 6*1024*1024; + + norm_tx_segment_size = segmentSize; + + TRACE("NormCaster::Start(%d, %d) ...\n", sender, receiver); + if (receiver) + { + if (!NormStartReceiver(norm_session, bufferSize)) + { + fprintf(stderr, "normCast error: unable to start NORM receiver\n"); + return false; + } + // Note: NormPreallocateRemoteSender() MUST be called AFTER NormStartReceiver() + NormPreallocateRemoteSender(norm_session, bufferSize, segmentSize, blockSize, numParity, bufferSize); + NormSetRxSocketBuffer(norm_session, rxSockBufferSize); + fprintf(stderr, "normCast: receiver ready.\n"); + } + if (sender) + { + if (norm_acking) + { + // ack-based flow control enabled on command-line, + // so disable timer-based flow control + NormSetFlowControl(norm_session, 0.0); + } + else + { + // Uncommant and adjust this for more/less robust timer-based flow control as desired + //NormSetFlowControl(norm_session, 10.0); + } + NormSetGrttEstimate(norm_session, 0.001); + //NormSetGrttMax(norm_session, 0.100); + NormSetBackoffFactor(norm_session, 0); + + // Pick a random instance id for now + struct timeval currentTime; + gettimeofday(¤tTime, NULL); + srand(currentTime.tv_usec); // seed random number generator + NormSessionId instanceId = (NormSessionId)rand(); + if (!NormStartSender(norm_session, instanceId, bufferSize, segmentSize, blockSize, numParity)) + { + fprintf(stderr, "normCast error: unable to start NORM sender\n"); + if (receiver) NormStopReceiver(norm_session); + return false; + } + //NormSetAutoParity(norm_session, 2); + NormSetTxSocketBuffer(norm_session, txSockBufferSize); + } + is_running = true; + return true; +} // end NormCaster::Start(); + +bool NormCaster::TxReady() const +{ + // This returns true if new tx data can be enqueued to NORM + // This is based on the state with respect to prior successful file + // enqueuing (or stream writing) and NORM_TX_QUEUE_EMPTY or + // NORM_TX_QUEUE_VACANCY notifications (tracked by the "norm_tx_vacancy" variable, + // _and_ (if ack-based flow control is enabled) the norm_tx_queue_count or + // norm_stream_buffer_count status. + if (norm_tx_vacancy) + { + if (norm_tx_queue_count >= norm_tx_queue_max) + return false; // still waiting for ACK + return true; + } + else + { + return false; + } +} // end NormCaster::TxReady() + +bool NormCaster::StageNextTxFile() +{ + // Pull next file path from "tx_file_list" as stage as "tx_pending_path" + if (tx_file_iterator.GetNextFile(tx_pending_path)) + { + // This code omits the source directory prefix from the name transmitted + // (Note that subdirectory structures will still be conveyed) + // The "tx_pending_prefix_len" indicates how many chararacters to skip + const ProtoFile::Path* pathParent = tx_file_iterator.GetCurrentPathItem(); + ASSERT(NULL != pathParent); + if (pathParent->IsDirectory()) + { + const char* prefix = pathParent->GetPath(); + tx_pending_prefix_len = strlen(prefix); + if (PROTO_PATH_DELIMITER != prefix[tx_pending_prefix_len - 1]) + tx_pending_prefix_len += 1; + } + else + { + tx_pending_prefix_len = 0; + } + return true; + } + else + { + // Either done need to reset tx_file_iterator + tx_pending_path[0] = '\0'; + tx_pending_prefix_len = 0; + return false; + } +} // end NormCaster::StageNextTxFile() + +void NormCaster::SendFiles() +{ + // Send pending file and subsequent files from tx_file_iterator as available/possible + while (TxFilePending() && TxReady()) + { + // Note EnqueueFileObject() may be subject to NORM flow control. + // Will be cued to retry via NORM_TX_QUEUE_VACANCY, NORM_TX_QUEUE_EMPTY, + // or NORM_TX_WATERMARK_COMPLETED (when ACK is used) notification. + if (EnqueueFileObject()) + { + sent_count++; + // Get next file name from our "tx_file_list" + if (!!StageNextTxFile()) + { + // We have reach end of tx_file_list, so either + // we're done or we reset tx_file_iterator + + // If we're done and requesting ACK, finish up nicely with final waterrmark + if (norm_acking) + { + if (NORM_OBJECT_INVALID == norm_last_object) + { + TRACE("normCast sender STOP 2 ...\n"); + is_running = false; // everything already sent and acked + } + else if (!norm_flushing) + { + NormSetWatermark(norm_session, norm_last_object, true); + } + } + } + } + else + { + // locked by NORM flow control ... + ASSERT(!TxReady()); + break; + } + } +} // end NormCaster::SendFiles() + +bool NormCaster::EnqueueFileObject() +{ + if (norm_acking) + { + assert(norm_tx_queue_count < norm_tx_queue_max); + if (norm_tx_queue_count >= norm_tx_queue_max) + return false; + } + + // This is our cheesy approach of using the NORM_INFO to convey the file name + // (TBD - implement something more sophisticated like our FCAST (RFC 6968) concept) + + unsigned int nameLen = strlen(tx_pending_path) - tx_pending_prefix_len; + const char* namePtr = tx_pending_path + tx_pending_prefix_len; + if (nameLen > norm_tx_segment_size) + { + fprintf(stderr, "normCast error: transmit file path \"%s\" exceeds NORM segment size limit!\n", namePtr); + nameLen = norm_tx_segment_size; + // TBD - refactor file name to preserve extension? + } + char nameInfo[PATH_MAX + 1]; + strncpy(nameInfo, namePtr, nameLen); + // Normalize path delimiters to '/' for transfer + for (int i = 0; i < nameLen; i++) + { + if (PROTO_PATH_DELIMITER == nameInfo[i]) + nameInfo[i] = '/'; + } + + // Enqueue the file for transmission + NormObjectHandle object = NormFileEnqueue(norm_session, tx_pending_path, nameInfo, nameLen); + if (NORM_OBJECT_INVALID == object) + { + // This might happen if a non-acking receiver is present and + // has nacked for the oldest object in the queue even if all + // of our acking receivers have acknowledged it. + //fprintf(stderr, "NO VACANCY count:%u max:%u\n", norm_tx_queue_count, norm_tx_queue_max); + norm_tx_vacancy = false; + return false; + } + fprintf(stderr, "normCast: enqueued \"%s\" for transmision ...\n", namePtr); + if (norm_acking) + { + // ack-based flow control has been enabled + norm_tx_queue_count++; + if (!norm_flow_control_pending && (norm_tx_queue_count >= (norm_tx_queue_max / 2))) + { + NormSetWatermark(norm_session, object, true); // overrideFlush == true + norm_last_object = object; + norm_flow_control_pending = true; + } + else if (norm_flushing) // per-message acking + { +#ifdef SHOOT_FIRST + NormSetWatermark(norm_session, object, true); + norm_last_object = object; +#else // ACK_LATER + if (norm_flow_control_pending) + { + norm_flush_object = object; // will be used as watermark upon flow control ack + } + else + { + NormSetWatermark(norm_session, object, true); + norm_last_object = object; + } +#endif // SHOOT_FIRST/ACK_LATER + } + else + { + norm_last_object = object; + } + TRACE("norm_last_object: %p\n", norm_last_object); + } + return true; +} // end NormCaster::EnqueueFileObject() + +void NormCaster::HandleNormEvent(const NormEvent& event) +{ + switch (event.type) + { + case NORM_TX_QUEUE_EMPTY: + case NORM_TX_QUEUE_VACANCY: + norm_tx_vacancy = true; + if (TxFilePending()) SendFiles(); + break; + + case NORM_GRTT_UPDATED: + //fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session)); + break; + + case NORM_TX_WATERMARK_COMPLETED: + if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) + { + //fprintf(stderr, "WATERMARK COMPLETED\n"); + norm_last_object = NORM_OBJECT_INVALID; + if (norm_flow_control_pending) + { + norm_tx_queue_count -= (norm_tx_queue_max / 2); + norm_flow_control_pending = false; + if (NORM_OBJECT_INVALID != norm_flush_object) + { + NormSetWatermark(norm_session, norm_flush_object, true); + norm_last_object = norm_flush_object; + norm_flush_object = NORM_OBJECT_INVALID; + } + } + if (NORM_OBJECT_INVALID == norm_last_object) + { + TRACE("normCast sender STOP 1 ...\n"); + is_running = false; + } + } + else + { + // TBD - we could see who did and how didn't ACK and possibly remove them + // from our acking "membership". For now, we are infinitely + // persistent by resetting watermark ack request without clearing + // flow control + if (NORM_OBJECT_INVALID == norm_flush_object) + { + NormResetWatermark(norm_session); + } + else // might as request ack for most recent enqueued object + { + NormSetWatermark(norm_session, norm_flush_object, true); + norm_last_object = norm_flush_object; + norm_flush_object = NORM_OBJECT_INVALID; + } + } + break; + + case NORM_TX_OBJECT_PURGED: + { + if(event.object == norm_flush_object) + norm_flush_object = NORM_OBJECT_INVALID; + // This is where we could delete the associated tx file if desired + // (e.g., for an "outbox" use case + break; + } + + case NORM_REMOTE_SENDER_INACTIVE: + //fprintf(stderr, "REMOTE SENDER INACTIVE node: %u\n", NormNodeGetId(event.sender)); + //NormNodeDelete(event.sender); + break; + + case NORM_RX_OBJECT_ABORTED: + //fprintf(stderr, "NORM_RX_OBJECT_ABORTED\n");// %hu\n", NormObjectGetTransportId(event.object)); + break; + + case NORM_RX_OBJECT_INFO: + { + // We use the NORM_INFO to contain the transferred file name + if (NORM_OBJECT_FILE != NormObjectGetType(event.object)) + { + fprintf(stderr, "normCast: received invalid object type?!\n"); + break; + } + // Rename rx file using newly received info + char fileName[PATH_MAX+1]; + fileName[PATH_MAX] = '\0'; + strncpy(fileName, rx_cache_path, PATH_MAX); + UINT16 pathLen = (UINT16)strlen(rx_cache_path); + pathLen = MIN(pathLen, PATH_MAX); + UINT16 len = NormObjectGetInfoLength(event.object); + len = MIN(len, (PATH_MAX - pathLen)); + NormObjectGetInfo(event.object, fileName + pathLen, len); + fileName[pathLen + len] = '\0'; + // Convert '/' in file info to directory delimiters + for (UINT16 i = pathLen; i < (pathLen+len); i++) + { + if ('/' == fileName[i]) + fileName[i] = PROTO_PATH_DELIMITER; + } + TRACE("got info renaming to \"%s\" ...\n", fileName); + if (!NormFileRename(event.object, fileName)) + perror("normCast: rx file rename error"); + break; + } + case NORM_RX_OBJECT_COMPLETED: + { + if (NORM_OBJECT_FILE != NormObjectGetType(event.object)) + { + fprintf(stderr, "normCast: received invalid object type?!\n"); + break; + } + char fileName[PATH_MAX + 1]; + fileName[PATH_MAX] = '\0'; + NormFileGetName(event.object, fileName, PATH_MAX); + fprintf(stderr, "normCast: completed reception of \"%s\"\n", fileName); + // TBD - implement receive post processing option ... + break; + } + + default: + break; + } + +} // end NormCaster::HandleNormEvent() + + +void Usage() +{ + fprintf(stderr, "Usage: normCast id {send &| recv }\n" + " [addr [/]][interface ][ack [,,...]\n" + " [cc|cce|ccl|rate ]\n" + " [debug ][trace]\n"); +} // end Usage() + +int main(int argc, char* argv[]) +{ + // REQUIRED parameters initiailization + NormNodeId nodeId = NORM_NODE_NONE; + bool send = false; + bool recv = false; + + char sessionAddr[64]; + strcpy(sessionAddr, "224.1.2.3"); + unsigned int sessionPort = 6003; + + NormNodeId ackingNodeList[256]; + unsigned int ackingNodeCount = 0; + bool flushing = false; + + double txRate = 0.0; // used for non-default NORM_FIXED ccMode + NormCaster::CCMode ccMode = NormCaster::NORM_CC; + const char* mcastIface = NULL; + + int debugLevel = 0; + const char* debugLog = NULL; // stderr by default + bool trace = false; + bool silentReceiver = false; + double txloss = 0.0; + bool loopback = false; + + NormCaster normCast; + + // Parse command-line + int i = 1; + while (i < argc) + { + const char* cmd = argv[i++]; + size_t len = strlen(cmd); + if (0 == strncmp(cmd, "send", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'send' file/dir argument ...\n"); + Usage(); + return -1; + } + // Use ProtoTokenator to process comma-delimited file/dir path items + ProtoTokenator tk(argv[i++], ','); + const char* path; + while (NULL != (path = tk.GetNextItem())) + { + // TBD - validate using ProtoFile::Exists() ??? + if (!normCast.AddTxItem(path)) + { + perror("normCast: 'send' error"); + Usage(); + return -1; + } + } + send = true; + } + else if (0 == strncmp(cmd, "recv", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'recv' !\n"); + Usage(); + return -1; + } + normCast.SetRxCacheDirectory(ptr); + recv = true; + } + else if (0 == strncmp(cmd, "loopback", len)) + { + loopback = true; + } + else if (0 == strncmp(cmd, "addr", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'addr[/port]' value!\n"); + Usage(); + return -1; + } + const char* addrPtr = argv[i++]; + const char* portPtr = strchr(addrPtr, '/'); + if (NULL == portPtr) + { + strncpy(sessionAddr, addrPtr, 63); + sessionAddr[63] = '\0'; + } + else + { + size_t addrLen = portPtr - addrPtr; + if (addrLen > 63) addrLen = 63; // should issue error message + strncpy(sessionAddr, addrPtr, addrLen); + sessionAddr[addrLen] = '\0'; + portPtr++; + sessionPort = atoi(portPtr); + } + } + else if (0 == strncmp(cmd, "id", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'id' value!\n"); + Usage(); + return -1; + } + nodeId = atoi(argv[i++]); + } + else if (0 == strncmp(cmd, "ack", len)) + { + // comma-delimited acking node id list + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'id' value!\n"); + Usage(); + return -1; + } + const char* alist = argv[i++]; + while ((NULL != alist) && (*alist != '\0')) + { + // TBD - Do we need to skip leading white space? + int id; + if (1 != sscanf(alist, "%d", &id)) + { + fprintf(stderr, "normCast error: invalid acking node list!\n"); + Usage(); + return -1; + } + ackingNodeList[ackingNodeCount] = NormNodeId(id); + ackingNodeCount++; + alist = strchr(alist, ','); + if (NULL != alist) alist++; // point past comma + } + } + else if (0 == strncmp(cmd, "flush", len)) + { + // "none", "passive", or "active" + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'flush' !\n"); + Usage(); + return -1; + } + const char* mode = argv[i++]; + if (0 == strcmp(mode, "none")) + { + flushing = false; + } + else if (0 == strcmp(mode, "passive")) + { + flushing = false; + } + else if (0 == strcmp(mode, "active")) + { + flushing = true; + } + else + { + fprintf(stderr, "normCast error: invalid 'flush' mode \"%s\"\n", mode); + return -1; + } + } + else if (0 == strncmp(cmd, "rate", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'rate' value!\n"); + Usage(); + return -1; + } + if (1 != sscanf(argv[i++], "%lf", &txRate)) + { + fprintf(stderr, "normCast error: invalid transmit rate!\n"); + Usage(); + return -1; + } + // set fixed-rate operation + ccMode = NormCaster::NORM_FIXED; + } + else if (0 == strcmp(cmd, "cc")) + { + ccMode = NormCaster::NORM_CC; + } + else if (0 == strcmp(cmd, "cce")) + { + ccMode = NormCaster::NORM_CCE; + } + else if (0 == strcmp(cmd, "ccl")) + { + ccMode = NormCaster::NORM_CCL; + } + else if (0 == strncmp(cmd, "interface", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'interface' !\n"); + Usage(); + return -1; + } + mcastIface = argv[i++]; + } + else if (0 == strncmp(cmd, "silent", len)) + { + silentReceiver = true; + } + else if (0 == strncmp(cmd, "txloss", len)) + { + if (1 != sscanf(argv[i++], "%lf", &txloss)) + { + fprintf(stderr, "normCast error: invalid 'txloss' value!\n"); + Usage(); + return -1; + } + } + else if (0 == strncmp(cmd, "debug", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'debug' !\n"); + Usage(); + return -1; + } + debugLevel = atoi(argv[i++]); + } + else if (0 == strncmp(cmd, "log", len)) + { + if (i >= argc) + { + fprintf(stderr, "normCast error: missing 'log' !\n"); + Usage(); + return -1; + } + debugLog = argv[i++]; + } + else if (0 == strncmp(cmd, "trace", len)) + { + trace = true; + } + else if (0 == strncmp(cmd, "help", len)) + { + Usage(); + return 0; + } + else + { + fprintf(stderr, "normCast error: invalid command \"%s\"!\n", cmd); + Usage(); + return -1; + } + } + + if (!send && !recv) + { + fprintf(stderr, "normCast error: not configured to send or recv!\n"); + Usage(); + return -1; + } + if (NORM_NODE_NONE == nodeId) + { + fprintf(stderr, "normCast error: no local 'id' provided!\n"); + Usage(); + return -1; + } + + // TBD - should provide more error checking of calls + NormInstanceHandle normInstance = NormCreateInstance(); + NormSetDebugLevel(debugLevel); + if (NULL != debugLog) + NormOpenDebugLog(normInstance, debugLog); + + // TBD - enhance NORM to support per-session or perhaps per-sender rx cache directories? + if (recv) + NormSetCacheDirectory(normInstance, normCast.GetRxCacheDirectory()); + + normCast.SetLoopback(loopback); + normCast.SetFlushing(flushing); + + if (!normCast.OpenNormSession(normInstance, sessionAddr, sessionPort, (NormNodeId)nodeId)) + { + fprintf(stderr, "normCast error: unable to open NORM session\n"); + NormDestroyInstance(normInstance); + return -1; + } + + if (silentReceiver) normCast.SetSilentReceiver(true); + if (txloss > 0.0) normCast.SetTxLoss(txloss); + + for (unsigned int i = 0; i < ackingNodeCount; i++) + normCast.AddAckingNode(ackingNodeList[i]); + + normCast.SetNormCongestionControl(ccMode); + if (NormCaster::NORM_FIXED == ccMode) + normCast.SetNormTxRate(txRate); + if (NULL != mcastIface) + normCast.SetNormMulticastInterface(mcastIface); + + if (trace) normCast.SetNormMessageTrace(true); + + // TBD - set NORM session parameters + normCast.Start(send, recv); + + if (normCast.TxFilePending()) normCast.SendFiles(); + +#ifdef WIN32 + HANDLE hStdout = GetStdHandle(STD_OUTPUT_HANDLE); + Win32InputHandler inputHandler; + inputHandler.Open(); + HANDLE handleArray[2]; + handleArray[0] = NormGetDescriptor(instance); + bool inputNeeded = false; // set this to 'true', and uncomment next line for STDIN input notification + //handleArray[1] = inputHandler.GetEventHandle(); +#else + int normfd = NormGetDescriptor(normInstance); + int inputfd = fileno(stdin); // Note that one could get input from something besides 'stdin' if desired + fd_set fdsetInput; + FD_ZERO(&fdsetInput); +#endif // if/else WIN32/UNIX + + while (normCast.IsRunning()) + { + // TBD - could add "stdin" as a descriptor to monitor + // for run-time commands, etc + // (on Windows, would use the win32InputHandler code in norm/examples) + bool normEventPending = false; + bool inputEventPending = false; +#ifdef WIN32 + DWORD handleCount = inputNeeded ? 2 : 1; + DWORD waitStatus = + MsgWaitForMultipleObjectsEx(handleCount, // number of handles in array + handleArray, // object-handle array + INFINITE, // time-out interval + QS_ALLINPUT, // input-event type + 0); + if ((WAIT_OBJECT_0 <= waitStatus) && (waitStatus < (WAIT_OBJECT_0 + handleCount))) + { + if (0 == (waitStatus - WAIT_OBJECT_0)) + normEventPending = true; + else + inputEventPending = true; + } + else if (-1 == waitStatus) + { + perror("normClient: MsgWaitForMultipleObjectsEx() error"); + break; + } + else + { + // TBD - any other status we should handle? + // (e.g. WAIT_TIMEOUT, WAIT_ABANDONED or WAIT_IO_COMPLETION) + continue; // ignore for now + } +#else // UNIX + int maxfd = normfd; + FD_SET(normfd, &fdsetInput); + //FD_SET(inputfd, &fdsetInput); + //if (inputfd > maxfd) maxfd = inputfd; + if (maxfd >= 0) + { + struct timeval timeout; + timeout.tv_sec = 1; + timeout.tv_usec = 0; + int result = select(maxfd+1, &fdsetInput, NULL, NULL, NULL); //&timeout); + switch (result) + { + case -1: + switch (errno) + { + case EINTR: + case EAGAIN: + continue; + default: + perror("normCast select() error"); + // TBD - stop NormCaster + break; + } + break; + case 0: + // 'timeout' condition + // TBD - implement file transmit / repeat interval timer + // shouldn't occur for now (no timeout) + //fprintf(stderr, "normCast timeout ...\n"); + break; + default: + if (FD_ISSET(normfd, &fdsetInput)) normEventPending = true; + if (FD_ISSET(inputfd, &fdsetInput)) inputEventPending = true; + break; + } // end switch(result + } // end if (maxfd >= 0) +#endif // if/else WIN32/UNIX + + // if (inputEventPending) + // { + // TBD - could parse input for runtime commands (e.g., send ), etc + // } + if (normEventPending) + { + // Handle all pending NORM notification events + // (Setting waitForEvent=false lets us get all available events without blocking) + // (Otherwise, one could build a simple main() (or threaded) loop around NormGetNextEvent() if desired) + NormEvent event; + while (NormGetNextEvent(normInstance, &event, false)) + { + normCast.HandleNormEvent(event); + } + } + // Invoke any 'timeout' actions here + + } // end while(normCast.IsRunning() + + NormCloseDebugLog(normInstance); + + fprintf(stderr, "destroying session ...\n"); + normCast.CloseNormSession(); + + fprintf(stderr, "destroying instance ...\n"); + + NormDestroyInstance(normInstance); + + normCast.Destroy(); + + fprintf(stderr, "normCast exiting ...\n"); + +} // end main() +