diff --git a/CMakeLists.txt b/CMakeLists.txt index 250e85c..c0126ad 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -212,6 +212,7 @@ if(NORM_BUILD_EXAMPLES) normMsgr normStreamer normCast + chant normClient normServer #wintest diff --git a/examples/chant.cpp b/examples/chant.cpp new file mode 100755 index 0000000..8b28a99 --- /dev/null +++ b/examples/chant.cpp @@ -0,0 +1,1419 @@ + +#include "normApi.h" +#include "protoTree.h" +#include "protoAddress.h" + +#include // for printf(), etc +#include // for srand() +#include // for strrchr(), memset(), etc +#include // for gettimeofday() +#include // for htons() +#include // for, well, fnctl() +#include // obvious child +#include // embarrassingly obvious +#include // for read() + +const unsigned int MSG_SIZE_MAX = 8192; +const unsigned int CHAT_NAME_MAX = 32; + + +// Setting SHOOT_FIRST to non-zero means that an ACK request +// will be used to advance the acking "watermark" point +// with each message fully written to the transmit stream. +// The alternative "ack later" behavior waits to send a new +// ACK request until any pending flow control ACK requeset +// has completed. This latter approach favors throughput +// over timeliness of message delivery. I.e., lower data +// rate applications that are concerned with low-latency message +// delivery can potentially benefit from the "shoot first" +// behavior while very high throughput applications that want +// to "keep the pipe full as possible" can benefit from the +// "ack later" behavior. The difference between these behaviors, +// since ACK requests are cued for all messages when flow +// control is _not_ pending, is somewhat subtle and developers +// may want to assess both behaviors for their application. +// Additionally, limiting ACK request to flow control only is +// another possible approach as well as dynamically updating +// something like the "tx_stream_buffer_count" with each +// message ACK request initiated could be possible. The caution +// with the SHOOT_FIRST type strategies and high throughput is +// the application may end up "chasing" the ACK request until +// flow control buffer limits are reached and end up with +// "dead air" time. There are always tradeoffs! + +//#define SHOOT_FIRST 0 + + +class ChantClient : public ProtoTree::Item +{ + public: + ChantClient(NormNodeHandle nodeHandle); + ~ChantClient(); + + void RecvData(); + void SetRxStream(NormObjectHandle stream); + void SetChatName(const char* text); + const char* GetChatName() const {return chat_name;} + + private: + const char* GetKey() const {return ((const char*)&node_handle);} + unsigned int GetKeysize() const {return (sizeof(NormNodeHandle) << 3);} + + NormNodeHandle node_handle; + char chat_name[CHAT_NAME_MAX+1]; // from NORM_INFO + NormObjectHandle rx_stream; + bool msg_sync; + char rx_buffer[MSG_SIZE_MAX+1]; + unsigned int rx_index; + unsigned int rx_msg_length; + bool rx_ready; + +}; // end class ChantClient + +class ChantClientTable : public ProtoTreeTemplate +{ + public: + ChantClient* FindClient(NormNodeHandle nodeHandle) + {return (ChantClient*)Find((const char*)&nodeHandle, sizeof(nodeHandle) << 3);} +}; + +ChantClient::ChantClient(NormNodeHandle nodeHandle) + : node_handle(nodeHandle), rx_stream(NORM_OBJECT_INVALID), + msg_sync(false), rx_index(0), rx_msg_length(0), + rx_ready(false) +{ + strcpy(chat_name, ""); +} + +ChantClient::~ChantClient() +{ +} + +void ChantClient::SetChatName(const char* text) +{ + strncpy(chat_name, text, CHAT_NAME_MAX); + chat_name[CHAT_NAME_MAX] = '\0'; +} // end ChatClient::SetChatName() + +void ChantClient::SetRxStream(NormObjectHandle stream) +{ + if (rx_stream != stream) + { + rx_stream = stream; + msg_sync = false; + rx_index = rx_msg_length = 0; + rx_ready = true; + } +} // end ChantClient::SetRxStream + + +void ChantClient::RecvData() +{ + while (true) + { + if (!msg_sync) + { + msg_sync = NormStreamSeekMsgStart(rx_stream); + if (!msg_sync) + { + break; // wait for next NORM_RX_OBJECT_UPDATED to re-sync + } + } + unsigned int bytesWanted = MSG_SIZE_MAX - rx_index; + unsigned int bytesRead = bytesWanted; + if (!NormStreamRead(rx_stream, rx_buffer + rx_index, &bytesRead)) + { + // Stream broken (should _not_ happen if norm_acking flow control) + //fprintf(stderr, "chant error: BROKEN stream detected, re-syncing ...\n"); + msg_sync = false; + rx_index = rx_msg_length = 0; + continue; + } + // Scan new received data for new lines indicating end of messages, + // print out complete messages, and save any partial message received + bool msgReady; + unsigned int msgIndex = 0; + const char* ptr = rx_buffer + rx_index; + unsigned int dataLen = rx_index + bytesRead; + //fprintf(stderr, "idx:%u bytesRead:%u dataLen:%u msgIndex:%u\n", rx_index, bytesRead, dataLen, msgIndex); + unsigned int i = 0; + do + { + msgReady = false; + for (; i < bytesRead; i++) + { + rx_msg_length++; + if (('\n' == *ptr++) || (MSG_SIZE_MAX == rx_msg_length)) + { + msgReady = true; + break; + } + } + if (msgReady) + { + fprintf(stdout, "%s: ", chat_name); + rx_buffer[msgIndex + rx_msg_length-1] = '\0'; + fprintf(stdout, "%s\n", rx_buffer+msgIndex); + msgIndex += rx_msg_length; + dataLen -= rx_msg_length; + rx_msg_length = 0; + } + } while (msgReady); + // Move any remaining partial message to beginning of rx_buffer + memmove(rx_buffer, rx_buffer+msgIndex, dataLen); + rx_index = rx_msg_length = dataLen; + msgIndex = 0; + if (bytesRead != bytesWanted) + { + break; // didn't get all asked for, wait for next NORM_RX_OBJECT_UPDATED + } + } // end while(true) + +} // end ChantClient::RecvData() + +class ChantCommand +{ + public: + ChantCommand(); + ~ChantCommand(); + + // some day build these directly into NORM API + enum CCMode {NORM_FIXED, NORM_CC, NORM_CCE, NORM_CCL}; + + enum + { + MSG_HEADER_SIZE = 2, // Big Endian message length header size + MSG_SIZE_MAX = 65535 // (including length header) + }; + + void SetLoopback(bool state) + { + loopback = state; + if (NORM_SESSION_INVALID != norm_session) + { + NormSetMulticastLoopback(norm_session, state); + //NormSetLoopback(norm_session, state); // test code + } + } + + void SetChatName(const char* text) + { + strncpy(chat_name, text, CHAT_NAME_MAX); + chat_name[CHAT_NAME_MAX] = '\0'; + } + + void SetFtiInfo(bool state) + { + fti_info = state; + if (NORM_SESSION_INVALID != norm_session) + NormLimitObjectInfo(norm_session, state); + } + + int GetInputDescriptor() const + {return fileno(input_file);} + + bool OpenNormSession(NormInstanceHandle instance, + const char* addr, + unsigned short port, + NormNodeId nodeId); + void CloseNormSession(); + + void SetNormCongestionControl(CCMode ccMode); + void SetFlushMode(NormFlushMode flushMode) + {flush_mode = flushMode;} + 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 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); + acking_node_count++; + norm_acking = true; // invoke ack-based flow control + } + void SetAutoAck(bool enable) + { + auto_ack = enable; + norm_acking = enable; + } + + bool Start(); + void Stop() + {is_running = false;} + bool IsRunning() const + {return is_running;} + void HandleNormEvent(const NormEvent& event); + + // Sender methods + int GetInputFile() const + {return input_fd;} + void SetInputReady() + {input_ready = true;} + bool InputReady() const + {return input_ready;} + bool InputNeeded() const + {return input_needed;} + void ReadInput(); + bool TxPending() const + {return (!input_needed && (input_index < input_msg_length));} + bool TxReady() const + {return (tx_ready && (!norm_acking || (tx_stream_buffer_count < tx_stream_buffer_max)));} + void SendData(); + unsigned int WriteToStream(const char* buffer, unsigned int numBytes); + void FlushStream(bool eom, NormFlushMode flushMode); + + + // These can only be called post-OpenNormSession() + void SetSilentReceiver(bool state) + {NormSetSilentReceiver(norm_session, state);} + void SetTxLoss(double txloss) + {NormSetTxLoss(norm_session, txloss);} + void SetRxLoss(double rxloss) + {NormSetRxLoss(norm_session, rxloss);} + + void SetSegmentSize(unsigned short segmentSize) + {segment_size = segmentSize;} + void SetBlockSize(unsigned short blockSize) + {block_size = blockSize;} + void SetNumParity(unsigned short numParity) + {num_parity = numParity;} + void SetAutoParity(unsigned short autoParity) + {auto_parity = autoParity;} + + void SetStreamBufferSize(unsigned int value) + {stream_buffer_size = value;} + + void SetProbeTOS(UINT8 value) + {probe_tos = value;} + + private: + NormSessionHandle norm_session; + bool is_multicast; + UINT8 probe_tos; + bool loopback; + bool is_running; + + char chat_name[CHAT_NAME_MAX+1]; + + // State variables for reading input messages for transmission + FILE* input_file; + int input_fd; // stdin by default + bool input_ready; + bool input_needed; + char input_buffer[MSG_SIZE_MAX]; + unsigned int input_msg_length; + unsigned int input_index; + + NormObjectHandle tx_stream; + bool tx_ready; + unsigned int tx_stream_buffer_max; + unsigned int tx_stream_buffer_threshold; // flow control threshold + unsigned int tx_stream_buffer_count; + unsigned int tx_stream_bytes_remain; + bool tx_watermark_pending; + bool norm_acking; + bool auto_ack; + unsigned int acking_node_count; + bool tx_ack_pending; + NormFlushMode flush_mode; // TBD - allow for "none", "passive", "active" options + bool fti_info; + + // Receive stream and state variables for writing received messages to output + ChantClientTable client_table; + + + // These are some options mainly for testing purposes + //double tx_loss; + unsigned long input_byte_count; + unsigned long tx_byte_count; + + unsigned short segment_size; + unsigned short block_size; + unsigned short num_parity; + unsigned short auto_parity; + + unsigned long stream_buffer_size; +}; // end class ChantCommand + +ChantCommand::ChantCommand() + : norm_session(NORM_SESSION_INVALID), is_multicast(false), probe_tos(0), loopback(false), is_running(false), + input_file(stdin), input_fd(fileno(stdin)), input_ready(true), + input_needed(false), input_msg_length(0), input_index(0), + tx_stream (NORM_OBJECT_INVALID), tx_ready(true), + tx_stream_buffer_max(0), tx_stream_buffer_count(0), tx_stream_bytes_remain(0), tx_watermark_pending(false), + norm_acking(false), auto_ack(false), acking_node_count(0), tx_ack_pending(false), flush_mode(NORM_FLUSH_ACTIVE), + fti_info(false), input_byte_count(0), tx_byte_count(0), + segment_size(1398), block_size(64), num_parity(0), auto_parity(0), + stream_buffer_size(2*1024*1024) +{ + strcpy(chat_name, "???"); +} + +ChantCommand::~ChantCommand() +{ +} + + +bool ChantCommand::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, "chant error: unable to create NORM session\n"); + return false; + } + if (is_multicast) + { + if (loopback) + { + NormSetRxPortReuse(norm_session, true); + NormSetMulticastLoopback(norm_session, true); + } + } + //NormSetLoopback(norm_session, loopback); + + // Set some default parameters (maybe we should put parameter setting in Start()) + NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM); + + if (!is_multicast) + NormSetDefaultUnicastNack(norm_session, true); + + NormSetTxRobustFactor(norm_session, 20); + + NormSetGrttProbingTOS(norm_session, probe_tos); + + NormSetFragmentation(norm_session, true); // so that IP ID gets set for SMF DPD + + return true; +} // end ChantCommand::OpenNormSession() + +void ChantCommand::CloseNormSession() +{ + if (NORM_SESSION_INVALID == norm_session) return; + NormDestroySession(norm_session); + norm_session = NORM_SESSION_INVALID; +} // end ChantCommand::CloseNormSession() + +void ChantCommand::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: // "fixed" constant data rate + NormSetEcnSupport(norm_session, false, false, false); + break; + } + if (NORM_FIXED != ccMode) + NormSetCongestionControl(norm_session, true); + else + NormSetCongestionControl(norm_session, false); +} // end ChantCommand::SetNormCongestionControl() + +bool ChantCommand::Start() +{ + // Note the session NORM buffer size is set the same s stream_buffer_size + unsigned int bufferSize = stream_buffer_size; + if (!NormStartReceiver(norm_session, bufferSize)) + { + fprintf(stderr, "chant error: unable to start NORM receiver\n"); + return false; + } + NormSetGrttEstimate(norm_session, 0.001); + //NormSetGrttMax(norm_session, 0.100); + NormSetBackoffFactor(norm_session, 0); + if (norm_acking) + { + // ack-based flow control enabled on command-line, + // so disable timer-based flow control + NormSetFlowControl(norm_session, 0.0); + NormTrackingStatus trackingMode = auto_ack ? NORM_TRACK_RECEIVERS : NORM_TRACK_NONE; + NormSetAutoAckingNodes(norm_session, trackingMode); + if (auto_ack && (0 == acking_node_count)) + { + // This allows for the receiver(s) to start after the sender + // as the sender will persistently send ack requests until + // a receiver responds. + NormAddAckingNode(norm_session, NORM_NODE_NONE); + } + } + // 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(); + NormLimitObjectInfo(norm_session, fti_info); + if (!NormStartSender(norm_session, instanceId, bufferSize, segment_size, block_size, num_parity)) + { + fprintf(stderr, "chant error: unable to start NORM sender\n"); + NormStopReceiver(norm_session); + return false; + } + if (auto_parity > 0) + NormSetAutoParity(norm_session, auto_parity < num_parity ? auto_parity : num_parity); + if (NORM_OBJECT_INVALID == (tx_stream = NormStreamOpen(norm_session, stream_buffer_size, chat_name, strlen(chat_name)))) + { + fprintf(stderr, "chant error: unable to open NORM tx stream\n"); + NormStopSender(norm_session); + NormStopReceiver(norm_session); + return false; + } + tx_stream_buffer_max = NormGetStreamBufferSegmentCount(bufferSize, segment_size, block_size); + tx_stream_buffer_max -= block_size; // a little safety margin (perhaps not necessary) + tx_stream_buffer_threshold = tx_stream_buffer_max / 8; + tx_stream_buffer_count = 0; + tx_stream_bytes_remain = 0; + tx_watermark_pending = false; + tx_ack_pending = false; + tx_ready = true; + input_index = input_msg_length = 0; + input_needed = true; + input_ready = true; + is_running = true; + return true; +} // end ChantCommand::Start(); + + + +void ChantCommand::ReadInput() +{ + //NormSuspendInstance(NormGetInstance(norm_session)); + while (input_needed && input_ready) + { + ssize_t result = read(input_fd, input_buffer + input_index, 1); + if (result > 0) + { + input_msg_length++; + // scan new input for end-of-line character to denote text message end + if ('\n' == input_buffer[input_index++]) + { + if (input_msg_length > 1) + { + input_needed = false; + input_index = 0; + if (TxReady()) SendData(); + } + else + { + // ignore empty messages + input_index = input_msg_length = 0; + continue; + } + } + } + else if (0 == result) + { + // end-of-file reached, TBD - trigger final flushing and wrap-up + fprintf(stderr, "chant: input end-of-file detected ...\n"); + NormStreamClose(tx_stream, true); + if (norm_acking) + { + NormSetWatermark(norm_session, tx_stream, true); + tx_ack_pending = false; + } + } + else + { + switch (errno) + { + case EINTR: + continue; // interrupted, try again + case EAGAIN: + // input starved, wait for next notification + input_ready = false; + break; + default: + // TBD - handle this better + perror("chant error reading input"); + break; + } + break; + } + } // end while (input_needed && input_ready) + //NormResumeInstance(NormGetInstance(norm_session)); +} // end ChantCommand::ReadInput() + +void ChantCommand::SendData() +{ + while (TxReady() && !input_needed) + { + // Note WriteToStream() or FlushStream() will set "tx_ready" to + // false upon flow control thus negating TxReady() status + assert(input_index < input_msg_length); + assert(input_msg_length); + int result = WriteToStream(input_buffer + input_index, input_msg_length - input_index); + input_index += result; + if (input_index == input_msg_length) + { + + // Complete message was sent, so set eom and optionally flush + if (NORM_FLUSH_NONE != flush_mode) + FlushStream(true, flush_mode); + else + NormStreamMarkEom(tx_stream); + + if (input_byte_count > input_msg_length) + { + // Move unsent bytes beginning of + unsigned int unsentBytes = input_byte_count - input_msg_length; + memmove(input_buffer, input_buffer+input_msg_length, unsentBytes); + input_index = input_byte_count = input_msg_length = unsentBytes; + } + else + { + input_index = input_byte_count = input_msg_length = 0; + } + input_needed = true; + } + else + { + //fprintf(stderr, "SendData() impeded by flow control\n"); + } + } // end while (TxReady() && !input_needed) +} // end ChantCommand::SendData() + +unsigned int ChantCommand::WriteToStream(const char* buffer, unsigned int numBytes) +{ + unsigned int bytesWritten; + if (norm_acking) + { + // This method uses NormStreamWrite(), but limits writes by explicit ACK-based flow control status + if (tx_stream_buffer_count < tx_stream_buffer_max) + { + // 1) How many buffer bytes are available? + unsigned int bytesAvailable = segment_size * (tx_stream_buffer_max - tx_stream_buffer_count); + bytesAvailable -= tx_stream_bytes_remain; // unflushed segment portion + if (bytesAvailable < numBytes) numBytes = bytesAvailable; + assert(numBytes); + // 2) Write to the stream + bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes); + tx_byte_count += bytesWritten; + // 3) Update "tx_stream_buffer_count" accordingly + unsigned int totalBytes = bytesWritten + tx_stream_bytes_remain; + unsigned int numSegments = totalBytes / segment_size; + tx_stream_bytes_remain = totalBytes % segment_size; + tx_stream_buffer_count += numSegments; + + //assert(bytesWritten == numBytes); // this could fail if timer-based flow control is left enabled + // 3) Check if we need to issue a watermark ACK request? + if (!tx_watermark_pending && (tx_stream_buffer_count >= tx_stream_buffer_threshold)) + { + // Initiate flow control ACK request + //fprintf(stderr, "write-initiated flow control ACK REQUEST\n"); + NormSetWatermark(norm_session, tx_stream);//, true); + tx_watermark_pending = true; + tx_ack_pending = false; + } + } + else + { + fprintf(stderr, "chant: sender flow control limited\n"); + return 0; + } + } + else + { + bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes); + tx_byte_count += bytesWritten; + } + if (bytesWritten != numBytes) //NormStreamWrite() was (at least partially) blocked + { + //fprintf(stderr, "NormStreamWrite() blocked by flow control ...\n"); + tx_ready = false; + } + return bytesWritten; +} // end ChantCommand::WriteToStream() + +void ChantCommand::FlushStream(bool eom, NormFlushMode flushMode) +{ + if (norm_acking) + { + bool setWatermark = false; + if (0 != tx_stream_bytes_remain) + { + // The flush will force the runt segment out, so we increment our buffer usage count + // (and initiate flow control watermark ack request if buffer mid-point threshold exceeded + tx_stream_buffer_count++; + tx_stream_bytes_remain = 0; + if (!tx_watermark_pending && (tx_stream_buffer_count >= tx_stream_buffer_threshold)) + { + setWatermark = true; + tx_watermark_pending = true; + //fprintf(stderr, "flush-initiated flow control ACK REQUEST\n"); + } + } + // The check for "tx_watermark_pending" here prevents a new watermark + // ack request from being set until the pending flow control ack is + // received. This favors avoiding dead air time over saving "chattiness" + if (setWatermark) + { + // Flush passive since watermark will invoke active request + // (TBD - do non-acking nodes NACK to watermark when not ack target?) + NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE); + } + else if (tx_watermark_pending) + { + // Pre-existing pending flow control watermark ack request +#if SHOOT_FIRST + // Go ahead and set a fresh watermark + // TBD - not sure this mode works properly ... may need to + // keep track of unacknowledged byte count and decrement accordingly + // when ack arrives + NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE); + setWatermark = true; +#else // ACK_LATER + // Wait until flow control ACK is received before issuing another ACK request + NormStreamFlush(tx_stream, eom, flushMode); + tx_ack_pending = true; // will call NormSetWatermark() upon flow control ack completion +#endif + } + else + { + // Since we're acking, use active ack request in lieu of active flush + NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE); + setWatermark = true; + } + if (setWatermark) + { + NormSetWatermark(norm_session, tx_stream, true); + tx_ack_pending = false; + } + } + else + { + NormStreamFlush(tx_stream, eom, flushMode); + } +} // end ChantCommand::FlushStream() + + +void ChantCommand::HandleNormEvent(const NormEvent& event) +{ + switch (event.type) + { + case NORM_TX_QUEUE_EMPTY: + //TRACE("chant: flow control empty ...\n"); + tx_ready = true; + break; + case NORM_TX_QUEUE_VACANCY: + //TRACE("chant: flow control relieved ...\n"); + tx_ready = true; + break; + + case NORM_GRTT_UPDATED: + //fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session)); + break; + + case NORM_ACKING_NODE_NEW: + if (0 == acking_node_count) + NormRemoveAckingNode(event.session, NORM_NODE_NONE); + acking_node_count++; + break; + + case NORM_TX_WATERMARK_COMPLETED: + if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) + { + //fprintf(stderr, "WATERMARK COMPLETED\n"); + if (0 == acking_node_count) + { + // Keep probing until some receiver shows up + NormResetWatermark(norm_session); + } + else if (tx_watermark_pending) + { + // Flow control ack request was pending. + tx_watermark_pending = false; + tx_stream_buffer_count -= tx_stream_buffer_threshold; + //fprintf(stderr, "flow control ACK completed\n"); + if (tx_ack_pending) + { + NormSetWatermark(norm_session, tx_stream, true); + tx_ack_pending = false; + } + } + } + else + { + // TBD - we could see who didn't ACK and possibly remove them + // from our acking list. For now, we are infinitely + // persistent by always resetting the watermark ack request + // For example, an application could make a decision at this + // point, depending upon some count of ACK request failures + // to choose to remove a previously included receiver. + fprintf(stderr, "flow control watermark reset\n"); + if (tx_ack_pending) + { + // May as well advance the ack request point + NormSetWatermark(norm_session, tx_stream, true); + } + else + { + // Uncomment for _persistent_ request, but may block forward progress + //NormResetWatermark(norm_session); + } + } + break; + + case NORM_TX_OBJECT_PURGED: + // tx_stream graceful close completed + NormStopSender(norm_session); + tx_stream = NORM_OBJECT_INVALID; + if (client_table.IsEmpty()) Stop(); + 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_NEW: + { + ChantClient* client = client_table.FindClient(event.sender); + if (NULL == client) + { + if (NULL == (client = new ChantClient(event.sender))) + { + fprintf(stderr, "chant error: unable to allocate new remote client state!\n"); + break; + } + client_table.Insert(*client); + char addrBuffer[16]; + unsigned int addrLen = 16; + if (NormNodeGetAddress(event.sender, addrBuffer, &addrLen)) + { + ProtoAddress addr; + ProtoAddress::Type addrType = (4 == addrLen) ? ProtoAddress::IPv4 : ProtoAddress::IPv6; + addr.SetRawHostAddress(addrType, addrBuffer, addrLen); + char text[256]; + addr.GetHostString(text, 256); + client->SetChatName(text); + } + } + if (NORM_OBJECT_STREAM != NormObjectGetType(event.object)) + { + fprintf(stderr, "chant error: received non-stream object?!\n"); + break; + } + client->SetRxStream(event.object); + break; + } + case NORM_RX_OBJECT_INFO: + { + ChantClient* client = client_table.FindClient(event.sender); + if (NULL == client) + { + fprintf(stderr, "chant: NORM_RX_OBJECT_INFO new rx stream ...\n"); + if (NULL == (client = new ChantClient(event.sender))) + { + fprintf(stderr, "chant error: unable to allocate new remote client state!\n"); + break; + } + client_table.Insert(*client); + client->SetRxStream(event.object); + } + char text[CHAT_NAME_MAX+1]; + unsigned int infoLen = NormObjectGetInfo(event.object, text, CHAT_NAME_MAX+1); + if (infoLen > CHAT_NAME_MAX) infoLen = CHAT_NAME_MAX; + text[infoLen] = '\0'; + fprintf(stdout, "(%s is \"%s\")\n", client->GetChatName(), text); + client->SetChatName(text); + break; + } + + case NORM_RX_OBJECT_UPDATED: + { + ChantClient* client = client_table.FindClient(event.sender); + if (NULL == client) + { + if (NULL == (client = new ChantClient(event.sender))) + { + fprintf(stderr, "chant error: unable to allocate new remote client state!\n"); + break; + } + client_table.Insert(*client); + client->SetRxStream(event.object); + } + client->RecvData(); + break; + } + case NORM_RX_OBJECT_ABORTED: + //fprintf(stderr, "chant: NORM_RX_OBJECT_ABORTED\n"); + break; + + case NORM_RX_OBJECT_COMPLETED: + //fprintf(stderr, "chant: NORM_RX_OBJECT_COMPLETED\n"); + break; + + default: + break; + } + +} // end ChantCommand::HandleNormEvent() + +void Usage() +{ + fprintf(stderr, "Usage: chant [name ][id ][addr [/]]\n" + " [interface ] [loopback] [info]\n" + " [cc|cce|ccl|rate ]\n" + " [ack _auto_|[,,...]]\n" + " [flush {none|passive|active}]\n" + " [boost] [debug ] [trace]\n" + " [log ] [segment ] [block ]\n" + " [parity ] [auto ]\n" + " [streambuffer ][silent]\n" + " [txloss ] [rxloss ]\n"); +} // end Usage() + +void PrintHelp() +{ + fprintf(stderr, "\nHelp for chant:\n\n") ; + fprintf(stderr, + "The 'chant' application sends text messages read from STDIN and outputs received messages to STDOUT.\n" + "Key command line options are:\n\n" + " name -- specifies chat user name ('handle') to advertise\n" + " addr [/] -- specifies the network address over which to send/receive NORM protocol\n" + " interface -- Specifies the name of the network interface on which to conduct NORM protocol\n" + " (e.g., 'eth0')\n" + " rate -- sets fixed sender rate\n" + " [cc|cce|ccl] -- Enables optional NORM congestion control mode (overrides 'rate')\n" + " ack [auto|none|] -- Instructs sender to request positive acknowledgement from receiver nodes (auto or comma-delimited nodeId list)\n" + " id -- Specifies the node id for the local NORM instance\n" + " loopback -- Enables 'loopback' sessions on the same host machine. Required for multicast loopback.\n" + " flush [] -- Choose 'none', 'passive', or 'active' message stream flushing mode. If 'none',\n" + " NORM_DATA packets will always be packed with message content up to the full\n" + " segment size. If 'passive', short NORM_DATA packets will be sent to transmit\n" + " any messages as soon as possible. If 'active', NORM stream will be flushed\n" + " on a per-message basis as with 'passive' mode, but positive acknowledgment will\n" + " _also_ be requested if a list of acking receiver node ids has beeen provided.\n" + " streambuffer -- Specifies the size of the NORM stream buffer (optional).\n\n"); + Usage(); + +} // end PrintHelp() + +int main(int argc, char* argv[]) +{ + // REQUIRED parameters initiailization + NormNodeId nodeId = NORM_NODE_NONE; + + char sessionAddr[64]; + strcpy(sessionAddr, "224.1.2.3"); + unsigned int sessionPort = 6003; + + double txRate = 0.0; // used for non-default NORM_FIXED ccMode + ChantCommand::CCMode ccMode = ChantCommand::NORM_CC; + const char* mcastIface = NULL; + NormNodeId ackingNodeList[256]; + unsigned int ackingNodeCount = 0; + + bool loopback = false; + bool ftiInfo = false; + int debugLevel = 0; + bool trace = false; + const char* logFile = NULL; + bool silentReceiver = false; + double txloss = 0.0; + double rxloss = 0.0; + // TBD - set these defaults to reasonable values or just use ChantCommand constructor defaults + unsigned long streamBufferSize = 1*1024*1024; + + // Instantiate a ChantCommand and set default params + ChantCommand chant; + chant.SetFlushMode(NORM_FLUSH_ACTIVE); + + chant.SetFtiInfo(true); // set true by default to reduce per-packet overhead + chant.SetAutoAck(true); // more succinct flushing + + // Parse command-line + int i = 1; + while (i < argc) + { + const char* cmd = argv[i++]; + size_t len = strlen(cmd); + if (0 == strncmp(cmd, "help", len)) + { + PrintHelp() ; + exit(0); + } + else if (0 == strncmp(cmd, "name", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'name' value!\n"); + Usage(); + return -1; + } + chant.SetChatName(argv[i++]); + } + else if (0 == strncmp(cmd, "loopback", len)) + { + loopback = true; + } + else if (0 == strncmp(cmd, "info", len)) + { + ftiInfo = true; + } + else if (0 == strncmp(cmd, "addr", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant 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, "chant 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, "chant error: missing 'id' value!\n"); + Usage(); + return -1; + } + const char* alist = argv[i++]; + if (0 == strcmp(alist, "auto")) + { + chant.SetAutoAck(true); + } + else if (0 == strcmp(alist, "none")) + { + chant.SetAutoAck(false); + } + else + { + while ((NULL != alist) && (*alist != '\0')) + { + // TBD - Do we need to skip leading white space? + int id; + if (1 != sscanf(alist, "%d", &id)) + { + fprintf(stderr, "chant 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, "nodeMsgr error: missing 'flush' !\n"); + Usage(); + return -1; + } + const char* mode = argv[i++]; + if (0 == strcmp(mode, "none")) + { + chant.SetFlushMode(NORM_FLUSH_NONE); + } + else if (0 == strcmp(mode, "passive")) + { + chant.SetFlushMode(NORM_FLUSH_PASSIVE); + } + else if (0 == strcmp(mode, "active")) + { + chant.SetFlushMode(NORM_FLUSH_ACTIVE); + } + else + { + fprintf(stderr, "normMsgr error: invalid 'flush' mode \"%s\"\n", mode); + return -1; + } + } + else if (0 == strncmp(cmd, "rate", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'rate' value!\n"); + Usage(); + return -1; + } + if (1 != sscanf(argv[i++], "%lf", &txRate)) + { + fprintf(stderr, "chant error: invalid transmit rate!\n"); + Usage(); + return -1; + } + // set fixed-rate operation + ccMode = ChantCommand::NORM_FIXED; + } + else if (0 == strcmp(cmd, "cc")) + { + ccMode = ChantCommand::NORM_CC; + } + else if (0 == strcmp(cmd, "cce")) + { + ccMode = ChantCommand::NORM_CCE; + } + else if (0 == strcmp(cmd, "ccl")) + { + ccMode = ChantCommand::NORM_CCL; + } + else if (0 == strncmp(cmd, "interface", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'interface' !\n"); + Usage(); + return -1; + } + mcastIface = argv[i++]; + } + else if (0 == strncmp(cmd, "segment", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'segment' size!\n"); + Usage(); + return -1; + } + unsigned short value; + if (1 != sscanf(argv[i++], "%hu", &value)) + { + fprintf(stderr, "chant error: invalid 'segment' size!\n"); + Usage(); + return -1; + } + chant.SetSegmentSize(value); + } + else if (0 == strncmp(cmd, "block", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'block' size!\n"); + Usage(); + return -1; + } + unsigned short value; + if (1 != sscanf(argv[i++], "%hu", &value)) + { + fprintf(stderr, "chant error: invalid 'block' size!\n"); + Usage(); + return -1; + } + chant.SetBlockSize(value); + } + else if (0 == strncmp(cmd, "parity", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'parity' count!\n"); + Usage(); + return -1; + } + unsigned short value; + if (1 != sscanf(argv[i++], "%hu", &value)) + { + fprintf(stderr, "chant error: invalid 'parity' count!\n"); + Usage(); + return -1; + } + chant.SetNumParity(value); + } + else if (0 == strncmp(cmd, "auto", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'auto' parity count!\n"); + Usage(); + return -1; + } + unsigned short value; + if (1 != sscanf(argv[i++], "%hu", &value)) + { + fprintf(stderr, "chant error: invalid 'auto' parity count!\n"); + Usage(); + return -1; + } + chant.SetAutoParity(value); + } + else if (0 == strncmp(cmd, "streambuffer", len)) + { + unsigned long value = 0 ; + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'streambuffer' size!\n"); + Usage(); + return -1; + } + if (1 != sscanf(argv[i++], "%lu", &value)) + { + fprintf(stderr, "chant error: invalid 'streambuffer' size!\n"); + Usage(); + return -1; + } + streamBufferSize = value; + } + 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, "chant error: invalid 'txloss' value!\n"); + Usage(); + return -1; + } + } + else if (0 == strncmp(cmd, "rxloss", len)) + { + if (1 != sscanf(argv[i++], "%lf", &rxloss)) + { + fprintf(stderr, "chant error: invalid 'rxloss' value!\n"); + Usage(); + return -1; + } + } + else if (0 == strncmp(cmd, "debug", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'debug' !\n"); + Usage(); + return -1; + } + debugLevel = atoi(argv[i++]); + } + else if (0 == strncmp(cmd, "trace", len)) + { + trace = true; + } + else if (0 == strncmp(cmd, "log", len)) + { + if (i >= argc) + { + fprintf(stderr, "chant error: missing 'log' !\n"); + Usage(); + return -1; + } + logFile = argv[i++]; + } + else if (0 == strncmp(cmd, "help", len)) + { + Usage(); + return 0; + } + else + { + fprintf(stderr, "chant error: invalid command \"%s\"!\n", cmd); + Usage(); + return -1; + } + } + + + /* + if (NORM_NODE_NONE == nodeId) + { + fprintf(stderr, "chant error: no local 'id' provided!\n"); + Usage(); + return -1; + } + */ + + + // TBD - should provide more error checking of NORM API calls + NormInstanceHandle normInstance = NormCreateInstance(); + + NormSetDebugLevel(debugLevel); + if ((NULL != logFile) && !NormOpenDebugLog(normInstance, logFile)) + { + perror("chant error: unable to open log file"); + Usage(); + return -1; + } + chant.SetStreamBufferSize(streamBufferSize); + + chant.SetLoopback(loopback); + chant.SetFtiInfo(ftiInfo); + + if (!chant.OpenNormSession(normInstance, sessionAddr, sessionPort, (NormNodeId)nodeId)) + { + fprintf(stderr, "chant error: unable to open NORM session\n"); + NormDestroyInstance(normInstance); + return -1; + } + + if (silentReceiver) chant.SetSilentReceiver(true); + if (txloss > 0.0) chant.SetTxLoss(txloss); + if (rxloss > 0.0) chant.SetRxLoss(rxloss); + + for (unsigned int i = 0; i < ackingNodeCount; i++) + chant.AddAckingNode(ackingNodeList[i]); + + chant.SetNormCongestionControl(ccMode); + if (ChantCommand::NORM_FIXED == ccMode) + chant.SetNormTxRate(txRate); + if (NULL != mcastIface) + chant.SetNormMulticastInterface(mcastIface); + + if (trace) chant.SetNormMessageTrace(true); + + // TBD - set NORM session parameters + chant.Start(); + + // TBD - add WIN32 support using win32InputHandler code + // and MsgWaitForMultipleObjectsEx() instead of select() + + int normfd = NormGetDescriptor(normInstance); + // Get input/output descriptors and set to non-blocking i/o + int inputfd = chant.GetInputDescriptor(); + if (-1 == fcntl(inputfd, F_SETFL, fcntl(inputfd, F_GETFL, 0) | O_NONBLOCK)) + perror("chant: fcntl(inputfd, O_NONBLOCK) error"); + fd_set fdsetInput, fdsetOutput; + FD_ZERO(&fdsetInput); + FD_ZERO(&fdsetOutput); + + while (chant.IsRunning()) + { + int maxfd = -1; + bool wait = true; + bool waitOnNorm = false; + if (chant.InputNeeded()) + { + if (chant.InputReady()) + { + FD_CLR(inputfd, &fdsetInput); + wait = false; + } + else + { + FD_SET(inputfd, &fdsetInput); + if (inputfd > maxfd) maxfd = inputfd; + } + } + else + { + FD_CLR(inputfd, &fdsetInput); + } + if (chant.TxPending()) + { + if (chant.TxReady()) + wait = false; + } + waitOnNorm = true; // always looking for receive notifications + + if (waitOnNorm) + { + // we need to wait until NORM is tx_ready or rx_ready + FD_SET(normfd, &fdsetInput); + if (normfd > maxfd) maxfd = normfd; + } + else + { + FD_CLR(normfd, &fdsetInput); + } + if (wait) + { + int result = select(maxfd+1, &fdsetInput, NULL, NULL, NULL); + switch (result) + { + case -1: + switch (errno) + { + case EINTR: + case EAGAIN: + continue; + default: + perror("chant select() error"); + // TBD - stop ChantCommand + break; + } + break; + case 0: + // timeout + break; + default: + if (FD_ISSET(inputfd, &fdsetInput)) + chant.SetInputReady(); + break; + } + } + // We always clear out/handle pending NORM API events + // (to keep event queue from building up) + unsigned int eventCount = 0; + const unsigned int maxBurstCount = 50; + NormEvent event; + while ((eventCount < maxBurstCount) && (NormGetNextEvent(normInstance, &event, false))) + { + chant.HandleNormEvent(event); + eventCount += 1; + } + + // As a result of input/output ready or NORM notification events: + // 1) Read from input if needed and ready + if (chant.InputNeeded() && chant.InputReady()) + chant.ReadInput(); + // 2) Send any pending tx message + if (chant.TxPending() && chant.TxReady()) + chant.SendData(); + + } // end while(chant.IsRunning() + + fflush(stderr); + + NormDestroyInstance(normInstance); + + fprintf(stderr, "chant exiting ...\n"); + + return 0; + +} // end main() + diff --git a/examples/normStreamer.cpp b/examples/normStreamer.cpp index a2b3f9b..4c68de0 100755 --- a/examples/normStreamer.cpp +++ b/examples/normStreamer.cpp @@ -17,7 +17,7 @@ #include #endif // LINUX -const unsigned int LOOP_MAX = 100; +const unsigned int LOOP_MAX = 1; // Setting SHOOT_FIRST to non-zero means that an ACK request // will be used to advance the acking "watermark" point @@ -577,7 +577,7 @@ bool NormStreamer::OpenNormSession(NormInstanceHandle instance, const char* addr NormSetLoopback(norm_session, loopback); // Set some default parameters (maybe we should put parameter setting in Start()) - NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM); + NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_CURRENT); if (!is_multicast) NormSetDefaultUnicastNack(norm_session, true); @@ -711,7 +711,7 @@ bool NormStreamer::Start(bool sender, bool receiver) void NormStreamer::ReadInputSocket() { unsigned int loopCount = 0; - NormSuspendInstance(NormGetInstance(norm_session)); + //NormSuspendInstance(NormGetInstance(norm_session)); while (input_needed && input_ready && (loopCount < LOOP_MAX)) { loopCount++; @@ -739,7 +739,7 @@ void NormStreamer::ReadInputSocket() input_ready = false; } } - NormResumeInstance(NormGetInstance(norm_session)); + //NormResumeInstance(NormGetInstance(norm_session)); } // end NormStreamer::ReadInputSocket() void NormStreamer::ReadInput() @@ -748,7 +748,7 @@ void NormStreamer::ReadInput() // The loop count makes sure we don't spend too much time here // before going back to the main loop to handle NORM events, etc unsigned int loopCount = 0; - NormSuspendInstance(NormGetInstance(norm_session)); + //NormSuspendInstance(NormGetInstance(norm_session)); while (input_needed && input_ready && (loopCount < LOOP_MAX)) { loopCount++; @@ -766,7 +766,9 @@ void NormStreamer::ReadInput() assert(input_index < input_msg_length); numBytes = input_msg_length - input_index; } + TRACE("reading STDIN ...\n"); ssize_t result = read(input_fd, input_buffer + input_index, numBytes); + TRACE(" result: %d\n", (int)result); if (result > 0) { input_index += result; @@ -833,7 +835,7 @@ void NormStreamer::ReadInput() break; } } // end while (input_needed && input_ready) - NormResumeInstance(NormGetInstance(norm_session)); + //NormResumeInstance(NormGetInstance(norm_session)); } // end NormStreamer::ReadInput() void NormStreamer::SendData() @@ -1000,7 +1002,7 @@ void NormStreamer::RecvData() // before going back to the main loop to handle NORM events, etc unsigned int loopCount = 0; // Reads data from rx_stream to available output_buffer - NormSuspendInstance(NormGetInstance(norm_session)); + //NormSuspendInstance(NormGetInstance(norm_session)); while (rx_needed && rx_ready && (loopCount < LOOP_MAX)) { loopCount++; @@ -1065,7 +1067,7 @@ void NormStreamer::RecvData() WriteOutput(); } } - NormResumeInstance(NormGetInstance(norm_session)); + //NormResumeInstance(NormGetInstance(norm_session)); } // end NormStreamer::RecvData() @@ -1120,8 +1122,8 @@ void NormStreamer::WriteOutput() if (0 != output_bucket_depth) { // Debit output token bucket since it's active - if (result > output_bucket_count) - TRACE("result:%d output_bucket_count:%u\n", (int)result, output_bucket_count); + //if (result > output_bucket_count) + // TRACE("result:%d output_bucket_count:%u\n", (int)result, output_bucket_count); ASSERT(output_bucket_count >= result); output_bucket_count -= result; } @@ -1174,7 +1176,7 @@ void NormStreamer::HandleNormEvent(const NormEvent& event) break; case NORM_GRTT_UPDATED: - //fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session)); + fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session)); break; case NORM_ACKING_NODE_NEW: @@ -1453,7 +1455,7 @@ int main(int argc, char* argv[]) unsigned long outputSocketBufferSize = 0; // 6*1024*1024; unsigned long txSocketBufferSize = 0; // 6*1024*1024; unsigned long rxSocketBufferSize = 0; // 6*1024*1024; - unsigned long streamBufferSize = 1*1024*1024; + unsigned long streamBufferSize = 10*1024*1024; // Instantiate a NormStreamer and set default params NormStreamer normStreamer; @@ -2136,6 +2138,7 @@ int main(int argc, char* argv[]) int maxfd = -1; int fdMask = 0; bool waitOnNorm = false; + bool waitOnInput = false; double timeoutInterval = -1.0; if (send) { @@ -2151,6 +2154,7 @@ int main(int argc, char* argv[]) FD_SET(inputfd, &fdsetInput); if (inputfd > maxfd) maxfd = inputfd; fdMask |= 0x01; + waitOnInput = true; } } else @@ -2268,6 +2272,7 @@ int main(int argc, char* argv[]) timeout.tv_sec = timeout.tv_usec = 0; } #endif // if/else LINUX + //ßTRACE("waitOnNorm:%d waitOnInput:%d\n", waitOnNorm, waitOnInput); int result = select(maxfd+1, &fdsetInput, &fdsetOutput, NULL, timeoutPtr); switch (result) { @@ -2304,9 +2309,14 @@ int main(int argc, char* argv[]) } // We always clear out/handle pending NORM API events // (to keep event queue from building up) + unsigned int eventCount = 0; + const unsigned int maxBurstCount = 50; NormEvent event; - while (NormGetNextEvent(normInstance, &event, false)) + while ((eventCount < maxBurstCount) && (NormGetNextEvent(normInstance, &event, false))) + { normStreamer.HandleNormEvent(event); + eventCount += 1; + } struct timeval thisTime; gettimeofday(&thisTime, NULL); diff --git a/makefiles/Makefile.common b/makefiles/Makefile.common index 74bb045..fb18d78 100755 --- a/makefiles/Makefile.common +++ b/makefiles/Makefile.common @@ -207,6 +207,16 @@ normCastApp: $(CASTAPP_OBJ) libnorm.a $(LIBPROTO) mkdir -p ../bin cp $@ ../bin/$@ +# (chant) NORM command-line chat sender/receiver +CHANT_SRC = $(EXAMPLE)/chant.cpp +CHANT_OBJ = $(CHANT_SRC:.cpp=.o) + +chant: $(CHANT_OBJ) libnorm.a $(LIBPROTO) + $(CC) $(CFLAGS) -o $@ $(CHANT_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) + mkdir -p ../bin + cp $@ ../bin/$@ + + # These are the new "NormSocket" API extension examples SERVER_SRC = $(EXAMPLE)/normServer.cpp $(EXAMPLE)/normSocket.cpp SERVER_OBJ = $(SERVER_SRC:.cpp=.o) diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index c176143..1a2510b 100755 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -1637,12 +1637,15 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) bool presetStream = false; NormObject* obj = NULL; bool doInsert = true; + bool seen = false; switch (status) { case OBJ_PENDING: { if (NULL != (obj = rx_table.Find(objectId))) { + // This checks for an object that's been "seen" but did not + // include Object FTI information previously. if (0 == obj->GetSize().GetOffset()) { // It's a seen object for which are awaiting FTI @@ -1651,6 +1654,7 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) gotFTI = true; obj->SetNackingMode(default_nacking_mode); doInsert = false; + seen = true; // Intentionally pass through to case OBJ_NEW } else @@ -1759,7 +1763,7 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) if (GetFtiData(msg, ftiData) || session.GetPresetFtiData(ftiData)) gotFTI = true; } - if (gotFTI) + if (gotFTI || presetStream) // this assumes presetStream matches if !gotFTI (if we get conflicting info, an abort/resync will be forced) { if (presetStream || obj->RxOpen(ftiData.GetObjectSize(), @@ -1791,7 +1795,7 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) NormBlockId syncId = blockId; stream->Decrement(syncId, stream->GetPendingMaskSize() - 1); if ((stream->Compare(blockId, NormBlockId(0)) >= 0) && - (stream->Compare(syncId, NormBlockId(0)) <= 0)) + (stream->Compare(syncId, NormBlockId(0)) <= 0) && !seen) { // Assume we are "in-range" of sender initial stream startup syncId = NormBlockId(0); @@ -1835,8 +1839,8 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) (unsigned long)GetId(), (UINT16)objectId); if (!presetStream) DeleteObject(obj); obj = NULL; - } - } + } // end if/else (gotFTI : ((NormMsg::INFO != msgType) && msg.FlagIsSet(NormObjectMsg::FLAG_INFO))) + } // end if (NULL != obj) break; } case OBJ_COMPLETE: @@ -1875,7 +1879,7 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) completion_count++; } } - } + } // end (if (NULL != obj) switch (repair_boundary) { case BLOCK_BOUNDARY: diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index 6aa30f0..fdda3da 100755 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -2794,16 +2794,16 @@ NormStreamObject::NormStreamObject(class NormSession& theSession, NormStreamObject::~NormStreamObject() { - Close(); + Close(); tx_offset = write_offset = read_offset = 0; NormBlock* b; while ((b = stream_buffer.Find(stream_buffer.RangeLo()))) { stream_buffer.Remove(b); b->EmptyToPool(segment_pool); - block_pool.Put(b); + block_pool.Put(b); } - stream_buffer.Destroy(); + stream_buffer.Destroy(); segment_pool.Destroy(); block_pool.Destroy(); } @@ -2898,10 +2898,10 @@ bool NormStreamObject::Open(UINT32 bufferSize, // since our objects are exclusively read _or_ write read_init = true; - read_index.block = read_index.segment = read_index.offset = 0; + read_index.block = read_index.segment = read_index.offset = 0; write_index.block = write_index.segment = 0; tx_index.block = tx_index.segment = 0; - tx_offset = write_offset = read_offset = 0; + tx_offset = write_offset = read_offset = 0; write_vacancy = true; stream_sync = false; flush_pending = false; @@ -3028,7 +3028,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) else { // Stream broken - return false; + return false; } } else @@ -3073,22 +3073,23 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) //stream_next_id = blockId + pending_mask.GetSize(); stream_next_id = blockId; Increment(stream_next_id, pending_mask.GetSize()); - if (NULL != sender) + if ((NULL != sender) && read_init) { - if (read_init && (NormSenderNode::SYNC_CURRENT != sender->GetSyncPolicy())) + // This is a fresh rx stream, so init the read indices + PLOG(PL_DEBUG, "NormStreamObject::StreamUpdateStatus() syncing stream to blockId: %lu\n", + (unsigned long)blockId.GetValue()); + read_init = false; + read_index.block = blockId; + read_index.segment = 0; + read_index.offset = 0; + read_offset = 0; + sender->DecrementResyncCount(); // correction since stream sync here will falsely increment + if ((NormSenderNode::SYNC_CURRENT != sender->GetSyncPolicy()) && + (0 != blockId.GetValue())) { - // This is a fresh rx stream, so init the read indices - read_init = false; - PLOG(PL_DEBUG, "NormStreamObject::StreamUpdateStatus() syncing stream to blockId: %lu\n", - (unsigned long)blockId.GetValue()); - read_index.block = blockId; - if (0 != blockId.GetValue()) stream_broken = true; - read_index.segment = 0; - read_index.offset = 0; - read_offset = 0; - sender->DecrementResyncCount(); // correction since stream sync here will falsely increment + stream_broken = true; } - } + } // end if ((NULL != sender) && read_init) // Since we're doing a resync including "read_init", dump any buffered data // (TBD) this may not be necessary??? and is thus currently commented-out code @@ -3230,16 +3231,8 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, NormSegmentId segmentId, const char* segment) { - UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); - if (read_init) - { - read_init = false; - read_index.block = blockId; - read_index.segment = segmentId; - read_index.offset = 0; - read_offset = segmentOffset; - read_ready = true; - } + //UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); + ASSERT(!read_init); //if ((blockId < read_index.block) || if ((Compare(blockId, read_index.block) < 0) || @@ -3273,7 +3266,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, if (Compare(blockId, block->GetId()) < 0) { PLOG(PL_DEBUG, "NormStreamObject::WriteSegment() blockId too old!?\n"); - return false; + return false; } while (block->IsPending()) { diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index d02008e..2b18482 100755 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -2275,7 +2275,6 @@ void NormSession::TxSocketRecvHandler(ProtoSocket &theSocket, unsigned int msgLength = NormMsg::MAX_SIZE; while (true) { - if (theSocket.RecvFrom(msg.AccessBuffer(), msgLength, msg.AccessAddress())) @@ -2631,6 +2630,7 @@ void NormTrace(const struct timeval ¤tTime, time_t secs = (time_t)currentTime.tv_sec; struct tm *ct = gmtime(&secs); #endif // if/else _WIN32_WCE + PLOG(PL_ALWAYS, "trace>%02d:%02d:%02d.%06lu ", (int)ct->tm_hour, (int)ct->tm_min, (int)ct->tm_sec, (unsigned int)currentTime.tv_usec); PLOG(PL_ALWAYS, "node>%lu %s>%s/%hu ", (unsigned long)localId, status, addr.GetHostString(), addr.GetPort()); @@ -2741,12 +2741,13 @@ void NormTrace(const struct timeval ¤tTime, { PLOG(PL_ALWAYS, "inst>%hu ", instId); // look for NormCCFeedback extension + bool ccExt = false; NormHeaderExtension ext; while (msg.GetNextExtension(ext)) { if (NormHeaderExtension::CC_FEEDBACK == ext.GetType()) { - clrFlag = ((NormCCFeedbackExtension &)ext).CCFlagIsSet(NormCC::CLR); + ccExt = true; break; } } @@ -2775,6 +2776,23 @@ void NormTrace(const struct timeval ¤tTime, PLOG(PL_ALWAYS, "NACK "); // TBD - provide deeper NACK inspection? } + if (ccExt) + { + clrFlag = ((NormCCFeedbackExtension &)ext).CCFlagIsSet(NormCC::CLR); + // Print ccRtt (only valid if pcap file is from sender node) + double ccRtt = NormUnquantizeRtt(((NormCCFeedbackExtension &)ext).GetCCRtt()); + double ccLoss = NormUnquantizeLoss32(((NormCCFeedbackExtension &)ext).GetCCLoss32()); + PLOG(PL_ALWAYS, "ccRtt:%lf ccLoss:%lf ", ccRtt, ccLoss); + // Print locally measured rtt (only valid if pcap file is from sender node) + struct timeval grttResponse; + if (NormMsg::NACK == msgType) + static_cast(msg).GetGrttResponse(grttResponse); + else + static_cast(msg).GetGrttResponse(grttResponse); + + double rtt = ProtoTime::Delta(ProtoTime(currentTime), ProtoTime(grttResponse)); + PLOG(PL_ALWAYS, "rtt:%lf ", rtt); + } break; } @@ -4811,46 +4829,46 @@ bool NormSession::OnTxTimeout(ProtoTimer & /*theTimer*/) switch (SendMessage(*msg)) { - case MSG_SEND_OK: - if (tx_rate > 0.0) - tx_timer.SetInterval(GetTxInterval(msgLength, tx_rate)); - if (advertise_repairs) - { - advertise_repairs = false; - suppress_rate = -1.0; // reset cc feedback suppression rate - } - else - { - ReturnMessageToPool(msg); - } - // Pre-serve to allow pre-prompt for empty tx queue - // (TBD) do this in a better way ??? There is a slight chance - // that with this approach some new data may get pre-queued - // when an interim repair request should be serviced first - // instead ??? - //if (message_queue.IsEmpty() && IsSender()) Serve(); - return true; // reinstall tx_timer + case MSG_SEND_OK: + if (tx_rate > 0.0) + tx_timer.SetInterval(GetTxInterval(msgLength, tx_rate)); + if (advertise_repairs) + { + advertise_repairs = false; + suppress_rate = -1.0; // reset cc feedback suppression rate + } + else + { + ReturnMessageToPool(msg); + } + // Pre-serve to allow pre-prompt for empty tx queue + // (TBD) do this in a better way ??? There is a slight chance + // that with this approach some new data may get pre-queued + // when an interim repair request should be serviced first + // instead ??? + //if (message_queue.IsEmpty() && IsSender()) Serve(); + return true; // reinstall tx_timer - case MSG_SEND_BLOCKED: - // Message was not sent due to to EWOULDBLOCK, so we invoke async i/o output notification - if (!advertise_repairs) - message_queue.Prepend(msg); - if (tx_timer.IsActive()) - tx_timer.Deactivate(); - tx_socket->StartOutputNotification(); - return false; // since timer was deactivated + case MSG_SEND_BLOCKED: + // Message was not sent due to to EWOULDBLOCK, so we invoke async i/o output notification + if (!advertise_repairs) + message_queue.Prepend(msg); + if (tx_timer.IsActive()) + tx_timer.Deactivate(); + tx_socket->StartOutputNotification(); + return false; // since timer was deactivated - case MSG_SEND_FAILED: - // Message was not sent due to socket error (no route, etc), so so just timeout and try again - // (TBD - is there something smarter we should do) - if (!advertise_repairs) - message_queue.Prepend(msg); - if (tx_rate > 0.0) - tx_timer.SetInterval(GetTxInterval(msgLength, tx_rate)); - else if (0.0 == tx_timer.GetInterval()) - tx_timer.SetInterval(0.001); - return true; // timer will be reactivated - } + case MSG_SEND_FAILED: + // Message was not sent due to socket error (no route, etc), so so just timeout and try again + // (TBD - is there something smarter we should do) + if (!advertise_repairs) + message_queue.Prepend(msg); + if (tx_rate > 0.0) + tx_timer.SetInterval(GetTxInterval(msgLength, tx_rate)); + else if (0.0 == tx_timer.GetInterval()) + tx_timer.SetInterval(0.001); + return true; // timer will be reactivated + } } else { diff --git a/wscript b/wscript index c88cc16..ff29f95 100755 --- a/wscript +++ b/wscript @@ -231,6 +231,8 @@ def build(ctx): 'normStreamSend', 'normMsgr', 'normStreamer', + 'normCast', + 'chant', 'normClient', 'normServer', #'wintest' # Windows only (can uncomment on Windows)