diff --git a/VERSION.TXT b/VERSION.TXT index 7e918fc..0f9718d 100644 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -1,5 +1,23 @@ NORM Version History +Version 1.4b2 +============= + - Fixed bug where NormObjectCancel() usage could result in extra + "release" of reference to NormObject + (Thanks to Alan McKechnie) + - Fixed minor issue with "repeatcount" command of "normApp.cpp" + where only the abbreviated command "repeat" was accepted. + (Thanks to Deric Sullivan) + - Added NormSetTxRobustFactor(), NormSetDefaultRxRobustFactor(), and + NormNodeSetRobustFactor() API calls + - Fixed issue with CC_INACTIVE being prematurely declared and modified + NORM RxSocketHandler to read a maximum of 100 consecutive packets + from socket at a time (This makes sure that receivers service timeouts, + provide feedback more timely, etc) + (Thanks to Peter Lu) + - Fixed case when repair request for old NORM_OBJECT_STREAM segment could + lockup advancement of stream. + Version 1.4b1 ============= - Fixed condition where receivers kept state for remote senders "stuck" diff --git a/common/normApi.cpp b/common/normApi.cpp index d9671a9..4c668d2 100644 --- a/common/normApi.cpp +++ b/common/normApi.cpp @@ -94,7 +94,7 @@ class NormInstance : public NormController static NormInstance* GetInstanceFromNode(NormNodeHandle nodeHandle) { if (NORM_NODE_INVALID == nodeHandle) return ((NormInstance*)NULL); - NormSession& session = ((NormNode*)nodeHandle)->GetSession(); + NormSession& session = ((NormServerNode*)nodeHandle)->GetSession(); return static_cast(session.GetSessionMgr().GetController()); } static NormInstance* GetInstanceFromObject(NormObjectHandle objectHandle) @@ -1182,6 +1182,19 @@ void NormSetGroupSize(NormSessionHandle sessionHandle, } } // end NormSetGroupSize() +NORM_API_LINKAGE +void NormSetTxRobustFactor(NormSessionHandle sessionHandle, + int robustFactor) +{ + NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormSession* session = (NormSession*)sessionHandle; + session->SetTxRobustFactor(robustFactor); + instance->dispatcher.ResumeThread(); + } +} // end NormSetTxRobustFactor() + NORM_API_LINKAGE NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle, const char* fileName, @@ -1300,8 +1313,7 @@ unsigned int NormStreamWrite(NormObjectHandle streamHandle, { NormStreamObject* stream = static_cast((NormObject*)streamHandle); - if (stream) - result = stream->Write(buffer, numBytes, false); + result = stream->Write(buffer, numBytes, false); instance->dispatcher.ResumeThread(); } return result; @@ -1317,13 +1329,10 @@ void NormStreamFlush(NormObjectHandle streamHandle, { NormStreamObject* stream = static_cast((NormObject*)streamHandle); - if (stream) - { - NormStreamObject::FlushMode saveFlushMode = stream->GetFlushMode(); - stream->SetFlushMode((NormStreamObject::FlushMode)flushMode); - stream->Flush(eom); - stream->SetFlushMode(saveFlushMode); - } + NormStreamObject::FlushMode saveFlushMode = stream->GetFlushMode(); + stream->SetFlushMode((NormStreamObject::FlushMode)flushMode); + stream->Flush(eom); + stream->SetFlushMode(saveFlushMode); instance->dispatcher.ResumeThread(); } } // end NormStreamFlush() @@ -1410,6 +1419,18 @@ bool NormSetWatermark(NormSessionHandle sessionHandle, return result; } // end NormSetWatermark() +NORM_API_LINKAGE +void NormCancelWatermark(NormSessionHandle sessionHandle) +{ + NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormSession* session = (NormSession*)sessionHandle; + session->ServerCancelWatermark(); + instance->dispatcher.ResumeThread(); + } +} // end NormSetWatermark() + NORM_API_LINKAGE bool NormAddAckingNode(NormSessionHandle sessionHandle, NormNodeId nodeId) @@ -1571,11 +1592,41 @@ NORM_API_LINKAGE void NormNodeSetRepairBoundary(NormNodeHandle nodeHandle, NormRepairBoundary repairBoundary) { - NormServerNode* node = static_cast((NormNode*)nodeHandle); + NormServerNode* node = static_cast((NormServerNode*)nodeHandle); if (node) node->SetRepairBoundary((NormServerNode::RepairBoundary)repairBoundary); } // end NormNodeSetRepairBoundary() + +NORM_API_LINKAGE +void NormSetDefaultRxRobustFactor(NormSessionHandle sessionHandle, + int robustFactor) +{ + NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormSession* session = (NormSession*)sessionHandle; + session->SetRxRobustFactor(robustFactor); + instance->dispatcher.ResumeThread(); + } +} // end NormSetDefaultRxRobustFactor() + +NORM_API_LINKAGE +void NormNodeSetRxRobustFactor(NormNodeHandle nodeHandle, + int robustFactor) +{ + if (NORM_NODE_INVALID != nodeHandle) + { + NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormServerNode* node = (NormServerNode*)nodeHandle; + node->SetRobustFactor(robustFactor); + instance->dispatcher.ResumeThread(); + } + } +} // end NormNodeSetRxRobustFactor() + NORM_API_LINKAGE bool NormStreamRead(NormObjectHandle streamHandle, char* buffer, @@ -1801,8 +1852,8 @@ NormNodeHandle NormObjectGetSender(NormObjectHandle objectHandle) NORM_API_LINKAGE NormNodeId NormNodeGetId(NormNodeHandle nodeHandle) { - NormNode* node = (NormNode*)nodeHandle; - if (node) + NormServerNode* node = (NormServerNode*)nodeHandle; + if (NULL != node) return node->GetId(); else return NORM_NODE_NONE; @@ -1820,7 +1871,7 @@ bool NormNodeGetAddress(NormNodeHandle nodeHandle, NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle); if (instance && instance->dispatcher.SuspendThread()) { - NormNode* node = (NormNode*)nodeHandle; + NormServerNode* node = (NormServerNode*)nodeHandle; const ProtoAddress& nodeAddr = node->GetAddress(); unsigned int addrLen = nodeAddr.GetLength(); if (addrBuffer && bufferLen && (addrLen <= *bufferLen)) diff --git a/common/normApi.h b/common/normApi.h index c2b2cbe..a9ca6f5 100644 --- a/common/normApi.h +++ b/common/normApi.h @@ -314,6 +314,10 @@ NORM_API_LINKAGE void NormSetGroupSize(NormSessionHandle sessionHandle, unsigned int groupSize); +NORM_API_LINKAGE +void NormSetTxRobustFactor(NormSessionHandle sessionHandle, + int robustFactor); + NORM_API_LINKAGE NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle, const char* fileName, @@ -365,6 +369,9 @@ NORM_API_LINKAGE bool NormSetWatermark(NormSessionHandle sessionHandle, NormObjectHandle objectHandle); +NORM_API_LINKAGE +void NormCancelWatermark(NormSessionHandle sessionHandle); + NORM_API_LINKAGE bool NormAddAckingNode(NormSessionHandle sessionHandle, NormNodeId nodeId); @@ -423,6 +430,14 @@ NORM_API_LINKAGE void NormNodeSetRepairBoundary(NormNodeHandle nodeHandle, NormRepairBoundary repairBoundary); +NORM_API_LINKAGE +void NormSetDefaultRxRobustFactor(NormSessionHandle sessionHandle, + int robustFactor); + +NORM_API_LINKAGE +void NormNodeSetRxRobustFactor(NormNodeHandle nodeHandle, + int robustFactor); + NORM_API_LINKAGE bool NormStreamRead(NormObjectHandle streamHandle, char* buffer, diff --git a/common/normApp.cpp b/common/normApp.cpp index 24c20f6..d8ab881 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -669,7 +669,7 @@ bool NormApp::OnCommand(const char* cmd, const char* val) return false; } } - else if (!strncmp("repeat", cmd, len)) + else if (!strncmp("repeatcount", cmd, len)) { tx_repeat_count = atoi(val); } diff --git a/common/normMessage.h b/common/normMessage.h index e750ecf..73de2b3 100644 --- a/common/normMessage.h +++ b/common/normMessage.h @@ -21,8 +21,6 @@ const UINT8 NORM_PROTOCOL_VERSION = 1; -const int NORM_ROBUST_FACTOR = 20; // default robust factor - // This value is used in a couple places in the code as // a safety check where some critical timeouts may be // less than expected operating system clock resolution @@ -268,8 +266,8 @@ class NormHeaderExtension protected: enum { - TYPE_OFFSET = 0, - LENGTH_OFFSET = TYPE_OFFSET + 1 + TYPE_OFFSET = 0, // UINT8 offset + LENGTH_OFFSET = TYPE_OFFSET + 1 // UINT8 offset }; UINT32* buffer; }; // end class NormHeaderExtension @@ -761,7 +759,7 @@ class NormCmdMsg : public NormMsg enum { INSTANCE_ID_OFFSET = MSG_OFFSET/2, - GRTT_OFFSET = (INSTANCE_ID_OFFSET*2)+2, + GRTT_OFFSET = (INSTANCE_ID_OFFSET+1)*2, BACKOFF_OFFSET = GRTT_OFFSET + 1, GSIZE_OFFSET = BACKOFF_OFFSET, FLAVOR_OFFSET = GSIZE_OFFSET + 1 @@ -1435,7 +1433,8 @@ class NormNackMsg : public NormMsg grttResponse.tv_usec = ntohl(buffer[GRTT_RESPONSE_USEC_OFFSET]); } //char* AccessRepairContent() {return (buffer + header_length);} - const UINT32* GetRepairContent() const {return (buffer + header_length/4);} + const UINT32* GetRepairContent() const + {return (buffer + header_length/4);} UINT16 GetRepairContentLength() const {return ((length > header_length) ? length - header_length : 0);} UINT16 UnpackRepairRequest(NormRepairRequest& request, diff --git a/common/normNode.cpp b/common/normNode.cpp index 29fcb99..40d52ba 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -54,8 +54,6 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId) nominal_packet_size(0), recv_total(0), recv_goodput(0), resync_count(0), nack_count(0), suppress_count(0), completion_count(0), failure_count(0) { - - repair_boundary = session.ClientGetDefaultRepairBoundary(); default_nacking_mode = session.ClientGetDefaultNackingMode(); unicast_nacks = session.ClientGetUnicastNacks(); @@ -66,10 +64,10 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId) repair_timer.SetRepeat(1); activity_timer.SetListener(this, &NormServerNode::OnActivityTimeout); - double activityInterval = 2*NormSession::DEFAULT_GRTT_ESTIMATE*NORM_ROBUST_FACTOR; + double activityInterval = 2*NormSession::DEFAULT_GRTT_ESTIMATE*session.GetTxRobustFactor(); if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN; activity_timer.SetInterval(activityInterval); - activity_timer.SetRepeat(NORM_ROBUST_FACTOR); + activity_timer.SetRepeat(robust_factor); cc_timer.SetListener(this, &NormServerNode::OnCCTimeout); cc_timer.SetInterval(0.0); @@ -308,6 +306,18 @@ void NormServerNode::FreeBuffers() segment_size = ndata = nparity = 0; } // end NormServerNode::FreeBuffers() +void NormServerNode::SetRobustFactor(int value) +{ + robust_factor = value; + // activity timer depends upon robust_factor + // (TBD) do a proper rescaling here instead? + double activityInterval = 2*session.GetTxRobustFactor()*grtt_estimate; + if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN; + activity_timer.SetInterval(activityInterval); + activity_timer.SetRepeat(robust_factor); + if (activity_timer.IsActive()) activity_timer.Reschedule(); +} // end NormServerNode::SetRobustFactor() + void NormServerNode::UpdateGrttEstimate(UINT8 grttQuantized) { grtt_quantized = grttQuantized; @@ -316,7 +326,7 @@ void NormServerNode::UpdateGrttEstimate(UINT8 grttQuantized) LocalNodeId(), GetId(), grtt_estimate); // activity timer depends upon sender's grtt estimate // (TBD) do a proper rescaling here instead? - double activityInterval = 2*NORM_ROBUST_FACTOR*grtt_estimate; + double activityInterval = 2*session.GetTxRobustFactor()*grtt_estimate; if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN; activity_timer.SetInterval(activityInterval); if (activity_timer.IsActive()) activity_timer.Reschedule(); @@ -418,6 +428,9 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime, // Respond immediately maxBackoff = 0.0; if (cc_timer.IsActive()) cc_timer.Deactivate(); + cc_timer.ResetRepeat(); + OnCCTimeout(cc_timer); + break; } else { @@ -456,6 +469,7 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime, DMSG(6, "NormServerNode::HandleCommand() node>%lu begin CC back-off: %lf sec)...\n", LocalNodeId(), backoffTime); session.ActivateTimer(cc_timer); + break; } // end if (CC_RATE == ext.GetType()) } // end while (GetNextExtension()) // Disable CC feedback if sender doesn't want it @@ -809,9 +823,11 @@ void NormServerNode::CalculateGrttResponse(const struct timeval& currentTime, void NormServerNode::DeleteObject(NormObject* obj) { if (rx_table.Remove(obj)) + { rx_pending_mask.Unset(obj->GetId()); - obj->Close(); - obj->Release(); + obj->Close(); + obj->Release(); + } } // end NormServerNode::DeleteObject() NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId blockId) @@ -1823,9 +1839,10 @@ void NormServerNode::Activate(bool isObjectMsg) { if (!activity_timer.IsActive()) { - double activityInterval = 2*NORM_ROBUST_FACTOR*grtt_estimate; + double activityInterval = 2*session.GetTxRobustFactor()*grtt_estimate; if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN; activity_timer.SetInterval(activityInterval); + activity_timer.SetRepeat(robust_factor); session.ActivateTimer(activity_timer); server_active = false; session.Notify(NormController::REMOTE_SENDER_ACTIVE, this, NULL); @@ -2087,7 +2104,7 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/) NormAckingNode::NormAckingNode(class NormSession& theSession, NormNodeId nodeId) - : NormNode(theSession, nodeId), ack_received(false), req_count(NORM_ROBUST_FACTOR) + : NormNode(theSession, nodeId), ack_received(false), req_count(theSession.GetTxRobustFactor()) { } diff --git a/common/normNode.h b/common/normNode.h index 2894be0..2dff03d 100644 --- a/common/normNode.h +++ b/common/normNode.h @@ -145,13 +145,14 @@ class NormAckingNode : public NormNode ~NormAckingNode(); bool IsPending() const {return (!ack_received && (req_count > 0));} - void Reset(unsigned int maxAttempts = NORM_ROBUST_FACTOR) + void Reset(unsigned int maxAttempts) { ack_received = false; req_count = maxAttempts; } void DecrementReqCount() {if (req_count > 0) req_count--;} - void ResetReqCount() {req_count = NORM_ROBUST_FACTOR;} + void ResetReqCount(unsigned int maxAttempts) + {req_count = maxAttempts;} unsigned int GetReqCount() const {return req_count;} bool AckReceived() const {return ack_received;} void MarkAckReceived() {ack_received = true;} @@ -225,6 +226,9 @@ class NormServerNode : public NormNode void SetDefaultNackingMode(NormObject::NackingMode nackingMode) {default_nacking_mode = nackingMode;} + // Should generally inherit NormSession::GetRxRobustFactor() + void SetRobustFactor(int value); + NormServerNode::RepairBoundary GetRepairBoundary() const {return repair_boundary;} // (TBD) force an appropriate RepairCheck on boundary change??? @@ -262,7 +266,6 @@ class NormServerNode : public NormNode bool BuffersAllocated() {return (0 != segment_size);} void FreeBuffers(); void Activate(bool isObjectMsg); - bool SyncTest(const NormObjectMsg& msg) const; void Sync(NormObjectId objectId); @@ -386,6 +389,7 @@ class NormServerNode : public NormNode void HandleRepairContent(const UINT32* buffer, UINT16 bufferLen); UINT16 instance_id; + int robust_factor; bool synchronized; NormObjectId sync_id; // only valid if(synchronized) NormObjectId next_id; // only valid if(synchronized) diff --git a/common/normObject.cpp b/common/normObject.cpp index 97befe8..5bd642a 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -374,10 +374,15 @@ bool NormObject::ActivateRepairs() NormBlock* block = block_buffer.Find(nextId); if (block) block->TxReset(GetBlockSize(nextId), nparity, autoParity, segment_size); // (TBD) This check can be eventually eliminated if everything else is done right - if (!pending_mask.Set(nextId)) + if (!pending_mask.Set(nextId)) + { + if (block) block->ClearPending(); DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error 1!\n", (UINT32)nextId); + } else + { repairsActivated = true; + } nextId++; } while (GetNextRepair(nextId)); repair_mask.Clear(); @@ -390,14 +395,18 @@ bool NormObject::ActivateRepairs() { if (block->ActivateRepairs(nparity)) { - //repairsActivated = true; DMSG(6, "NormObject::ActivateRepairs() node>%lu obj>%hu activated blk>%lu segment repairs ...\n", LocalNodeId(), (UINT16)transport_id, (UINT32)block->GetId()); // (TBD) This check can be eventually eliminated if everything else is done right if (!pending_mask.Set(block->GetId())) + { + block->ClearPending(); DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error 2!\n", (UINT32)block->GetId()); + } else + { repairsActivated = true; + } } } return repairsActivated; @@ -1525,7 +1534,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) NormBlockId blockId; if (!GetFirstPending(blockId)) { - // Attempt to advance stream (probably was repair-delayed) + // Attempt to advance stream (probably had been repair-delayed) if (IsStream()) { static_cast(this)->StreamAdvance(); @@ -1588,7 +1597,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) { NormBlock* b = block_buffer.Find(block_buffer.RangeLo()); ASSERT(NULL != b); - //ASSERT(!b->IsPending()); (TBD) ??? Why might "b" be pending? + ASSERT(!b->IsPending()); bool push = static_cast(this)->GetPushMode(); if (!push && (b->IsRepairPending() || IsRepairSet(b->GetId()))) { @@ -1609,11 +1618,12 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) } else if (IsStream()) { - DMSG(4, "NormObject::NextServerMsg() node>%lu Warning! can't repair old block\n", LocalNodeId()); + DMSG(3, "NormObject::NextServerMsg() node>%lu Warning! can't repair old stream block\n", LocalNodeId()); session.ServerPutFreeBlock(block); + repair_mask.Unset(blockId); // just in case pending_mask.Unset(blockId); // Unlock (set to non-pending status) the corresponding stream_buffer block - static_cast(this)->UnlockBlock(blockId); + static_cast(this)->UnlockBlock(blockId); return NextServerMsg(msg); } else @@ -1640,10 +1650,32 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) if (0 == payloadLength) { // (TBD) deal with read error - //(for streams, it currently means the stream is non-pending) - if (!IsStream()) + //(for streams, it currently means the stream is pending, but app hasn't yet written data) + if (!IsStream()) + { DMSG(0, "NormObject::NextServerMsg() ReadSegment() error\n"); - return false; + return false; + } + else if (static_cast(this)->IsOldBlock(blockId)) + { + DMSG(3, "NormObject::NextServerMsg() node>%lu Warning! can't repair old stream segment\n", LocalNodeId()); + block->UnsetPending(segmentId); + if (!block->IsPending()) + { + // End of old block reached + block->ResetParityCount(nparity); + pending_mask.Unset(blockId); + // for EMCON sending, mark NORM_INFO for re-transmission, if applicable + if (session.SndrEmcon() && HaveInfo()) + pending_info = true; + } + return NextServerMsg(msg); + } + else + { + // App hasn't written data for this block yet + return false; + } } data->SetPayloadLength(payloadLength); @@ -1688,7 +1720,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) if (!block->IsPending()) { // End of block reached - block->ResetParityCount(nparity); + block->ResetParityCount(nparity); pending_mask.Unset(blockId); // for EMCON sending, mark NORM_INFO for re-transmission, if applicable if (session.SndrEmcon() && HaveInfo()) @@ -1746,14 +1778,14 @@ void NormStreamObject::StreamAdvance() } else { - DMSG(4, "NormStreamObject::StreamAdvance() warning: node>%lu pending segment repairs (blk>%lu) " + DMSG(3, "NormStreamObject::StreamAdvance() warning: node>%lu pending segment repairs (blk>%lu) " "delaying stream advance ...\n", LocalNodeId(), (UINT32)block->GetId()); } } } else { - DMSG(4, "NormStreamObject::StreamAdvance() warning: node>%lu pending block repair delaying stream advance ...\n", + DMSG(3, "NormStreamObject::StreamAdvance() warning: node>%lu pending block repair delaying stream advance ...\n", LocalNodeId()); } } // end NormStreamObject::StreamAdvance() @@ -2469,7 +2501,8 @@ bool NormStreamObject::LockBlocks(NormBlockId firstId, NormBlockId lastId) NormBlock* block = stream_buffer.Find(nextId); if (block) { - block->SetPending(0, ndata); + UINT16 numData = GetBlockSize(nextId); + block->SetPending(0, numData); } else { @@ -2618,11 +2651,13 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, if (!block) { //DMSG(0, "NormStreamObject::ReadSegment() stream starved (1)\n"); + if (!stream_buffer.IsEmpty() && (blockId < stream_buffer.RangeLo())) + DMSG(0, "NormStreamObject::ReadSegment() error: attempted to read old block> %lu\n", (UINT32)blockId); + return false; } // (TBD) should we check to see if "blockId > write_index.block" ? - if ((blockId == write_index.block) && - (segmentId >= write_index.segment)) + if ((blockId == write_index.block) && (segmentId >= write_index.segment)) { //DMSG(0, "NormStreamObject::ReadSegment(blk>%lu seg>%hu) stream starved (2) (write_index>%lu:%hu)\n", // (UINT32)blockId, (UINT16)segmentId, (UINT32)write_index.block, (UINT16)write_index.segment); @@ -2850,7 +2885,7 @@ void NormStreamObject::Prune(NormBlockId blockId, bool updateStatus) { bool resync = false; NormBlock* block; - while ((block = block_buffer.Find(block_buffer.RangeLo()))) + while (NULL != (block = block_buffer.Find(block_buffer.RangeLo()))) { if (block->GetId() < blockId) { @@ -3469,8 +3504,7 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) } else { - //if (0 != nBytes) - session.TouchServer(); + session.TouchServer(); } return nBytes; } // end NormStreamObject::Write() diff --git a/common/normObject.h b/common/normObject.h index 0c74698..24ff445 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -401,6 +401,9 @@ class NormStreamObject : public NormObject } void SetPushMode(bool state) {push_mode = state;} bool GetPushMode() const {return push_mode;} + + bool IsOldBlock(NormBlockId blockId) const + {return (!stream_buffer.IsEmpty() && (blockId < stream_buffer.RangeLo()));} bool IsClosing() {return stream_closing;} bool HasVacancy() diff --git a/common/normSegment.h b/common/normSegment.h index c6066aa..5a4b3f8 100644 --- a/common/normSegment.h +++ b/common/normSegment.h @@ -245,8 +245,8 @@ class NormBlock int flags; UINT16 erasure_count; - UINT16 parity_count; - UINT16 parity_offset; + UINT16 parity_count; // how many fresh parity we are currently planning to send + UINT16 parity_offset; // offset from where our fresh parity will be sent UINT16 seg_size_max; ProtoBitmask pending_mask; diff --git a/common/normSession.cpp b/common/normSession.cpp index 1d91467..e95decb 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -16,20 +16,22 @@ const UINT16 NormSession::DEFAULT_NPARITY = 32; const UINT16 NormSession::DEFAULT_TX_CACHE_MIN = 8; const UINT16 NormSession::DEFAULT_TX_CACHE_MAX = 256; +const int NormSession::DEFAULT_ROBUST_FACTOR = 20; // default robust factor + NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) : session_mgr(sessionMgr), notify_pending(false), tx_port(0), tx_socket_actual(ProtoSocket::UDP), tx_socket(&tx_socket_actual), rx_socket(ProtoSocket::UDP), local_node_id(localNodeId), ttl(DEFAULT_TTL), tos(0), loopback(false), rx_port_reuse(false), rx_addr_bind(false), tx_rate(DEFAULT_TRANSMIT_RATE/8.0), tx_rate_min(-1.0), tx_rate_max(-1.0), - backoff_factor(DEFAULT_BACKOFF_FACTOR), is_server(false), instance_id(0), + backoff_factor(DEFAULT_BACKOFF_FACTOR), is_server(false), + tx_robust_factor(DEFAULT_ROBUST_FACTOR), instance_id(0), ndata(DEFAULT_NDATA), nparity(DEFAULT_NPARITY), auto_parity(0), extra_parity(0), sndr_emcon(false), encoder(NULL), next_tx_object_id(0), tx_cache_count_min(DEFAULT_TX_CACHE_MIN), tx_cache_count_max(DEFAULT_TX_CACHE_MAX), tx_cache_size_max((UINT32)20*1024*1024), - flush_count(NORM_ROBUST_FACTOR+1), posted_tx_queue_empty(false), acking_node_count(0), watermark_pending(false), tx_repair_pending(false), advertise_repairs(false), suppress_nonconfirmed(false), suppress_rate(-1.0), suppress_rtt(-1.0), @@ -41,7 +43,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) grtt_decrease_delay_count(DEFAULT_GRTT_DECREASE_DELAY), grtt_response(false), grtt_current_peak(0.0), grtt_age(0.0), cc_enable(false), cc_sequence(0), cc_slow_start(true), cc_active(false), - is_client(false), unicast_nacks(false), + is_client(false), rx_robust_factor(DEFAULT_ROBUST_FACTOR), unicast_nacks(false), client_silent(false), rcvr_ignore_info(false), rcvr_max_delay(-1), default_repair_boundary(NormServerNode::BLOCK_BOUNDARY), default_nacking_mode(NormObject::NACK_NORMAL), @@ -279,7 +281,6 @@ void NormSession::SetTxRateInternal(double txRate) else { tx_timer.Deactivate(); - } } else if ((0.0 == tx_rate) && IsOpen()) @@ -298,7 +299,6 @@ void NormSession::SetTxRateInternal(double txRate) grtt_quantized = NormQuantizeRtt(grtt_measured); grtt_advertised = NormUnquantizeRtt(grtt_quantized); - // What do we do when "pktInterval" > "grtt_max"? // We will take our lumps with some extra activity timeout NACKs when they happen? if (grtt_advertised > grtt_max) @@ -329,7 +329,6 @@ void NormSession::SetTxRateBounds(double rateMin, double rateMax) rateMax = temp; } } - if (rateMin < 0.0) tx_rate_min = -1.0; else if (rateMin < 8.0) @@ -437,7 +436,8 @@ bool NormSession::StartServer(UINT16 instanceId, ndata = numData; nparity = numParity; is_server = true; - flush_count = NORM_ROBUST_FACTOR+1; // (TBD) parameterize robust_factor + + flush_count = (GetTxRobustFactor() < 0) ? 0 : (GetTxRobustFactor() + 1); if (cc_enable) { @@ -559,8 +559,17 @@ void NormSession::Serve() obj = tx_table.Find(objectId); ASSERT(obj); } + + // (TBD) code to support app-defined commands will go here + /* + if (command_pending && !command_timer.IsActive()) + { + SenderQueueAppCommand() xxx + } + + */ + - // (TBD) Work on this whole watermark flush check logic! if (watermark_pending && !flush_timer.IsActive()) { // Determine next message (objectId::blockId::segmentId) to be sent @@ -621,47 +630,6 @@ void NormSession::Serve() (UINT16)watermark_object_id, (UINT32)watermark_block_id, (UINT16)watermark_segment_id, (UINT16)nextObjectId, (UINT32)nextBlockId, (UINT16)nextSegmentId); } - /* - - // Get index (objectId::blockId::segmentId) of next repair pending segment - // (This seems like alot of work to do for each call to NormSession::Serve(). Yuk! - // (well when a watermark is pending ... which could be alot of the time - // i.e. iterating through the tx_table, looking for repair pending objects ... - // (TBD) Is there a better way such as keeping state elsewhere on minimum - // pending repairs to save doing the work here?jo - NormObjectId repairObjectId = next_tx_object_id; - NormBlockId repairBlockId = 0; - NormSegmentId repairSegmentId = 0; - if (!ServerGetFirstRepairPending(repairObjectId)) - repairObjectId = next_tx_object_id; - NormObjectTable::Iterator iterator(tx_table); - while ((nextObj = iterator.GetNextObject())) - { - if (repairObjectId < nextObj->GetId()) - { - nextObj = tx_table.Find(repairObjectId); - break; - } - else if (nextObj->IsRepairPending()) - { - repairObjectId = nextObj->GetId(); - break; - } - } - if (nextObj) nextObj->FindRepairIndex(repairBlockId, repairSegmentId); - - // Get min of next transmit pending or repair pending segment - if ((repairObjectId < nextObjectId) || - ((repairObjectId == nextObjectId) && - ((repairBlockId < nextBlockId) || - ((repairBlockId == nextBlockId) && - (repairSegmentId < nextSegmentId))))) - { - nextObjectId = repairObjectId; - nextBlockId = repairBlockId; - nextSegmentId = repairSegmentId; - } - */ } // end if (tx_repair_pending) if ((nextObjectId > watermark_object_id) || @@ -670,6 +638,7 @@ void NormSession::Serve() ((nextBlockId == watermark_block_id) && (nextSegmentId > watermark_segment_id))))) { + // The sender tx position is > watermark if (ServerQueueWatermarkFlush()) { watermark_active = true; @@ -684,6 +653,7 @@ void NormSession::Serve() } else { + // The sender tx position is < watermark // Reset non-acked acking nodes since server has rewound if (watermark_active) { @@ -692,7 +662,7 @@ void NormSession::Serve() NormAckingNode* next; while ((next = static_cast(iterator.GetNextNode()))) { - next->ResetReqCount(); + next->ResetReqCount(GetTxRobustFactor()); } } } @@ -749,11 +719,11 @@ void NormSession::Serve() // Queue flush message if (!flush_timer.IsActive()) { - if (flush_count < NORM_ROBUST_FACTOR) + if ((GetTxRobustFactor() < 0) || (flush_count < GetTxRobustFactor())) { ServerQueueFlush(); } - else if (NORM_ROBUST_FACTOR == flush_count) + else if (GetTxRobustFactor() == flush_count) { DMSG(6, "NormSession::Serve() node>%lu server flush complete ...\n", LocalNodeId()); @@ -768,8 +738,9 @@ void NormSession::Serve() } } } - //ASSERT(stream->IsPending() || stream->IsRepairPending()); - if (!posted_tx_queue_empty && stream->IsPending() && !stream->IsClosing()) + //ASSERT(stream->IsPending() || stream->IsRepairPending() || stream->IsClosing()); + if (!posted_tx_queue_empty && !stream->IsClosing() && stream->IsPending()) + // post if pending || !repair_timer.IsActive() || (repair_timer.GetRepeatCount() == 0) ??? { //data_active = false; posted_tx_queue_empty = true; @@ -801,8 +772,8 @@ void NormSession::Serve() Notify(NormController::TX_QUEUE_EMPTY, (NormServerNode*)NULL, (NormObject*)NULL); // (TBD) Was session deleted? return; - } - if (flush_count < NORM_ROBUST_FACTOR) + } + if ((GetTxRobustFactor() < 0) || (flush_count < GetTxRobustFactor())) { // Queue flush message if (!tx_repair_pending) // don't queue flush if repair pending @@ -811,7 +782,7 @@ void NormSession::Serve() DMSG(8, "NormSession::Serve() node>%lu NORM_CMD(FLUSH) deferred by pending repairs ...\n", LocalNodeId()); } - else if (flush_count == NORM_ROBUST_FACTOR) + else if (GetTxRobustFactor() == flush_count) { DMSG(6, "NormSession::Serve() node>%lu server flush complete ...\n", LocalNodeId()); @@ -836,11 +807,17 @@ void NormSession::ServerSetWatermark(NormObjectId objectId, // Reset acking_node_list NormNodeTreeIterator iterator(acking_node_tree); NormNode* next; + int robustFactor = GetTxRobustFactor(); while ((next = iterator.GetNextNode())) - static_cast(next)->Reset(NORM_ROBUST_FACTOR); + static_cast(next)->Reset(robustFactor); PromptServer(); } // end Norm::ServerSetWatermark() +void NormSession::ServerCancelWatermark() +{ + watermark_pending = false; +} // end NormSession::ServerCancelWatermark() + bool NormSession::ServerAddAckingNode(NormNodeId nodeId) { NormAckingNode* theNode = static_cast(acking_node_tree.FindNodeById(nodeId)); @@ -849,7 +826,7 @@ bool NormSession::ServerAddAckingNode(NormNodeId nodeId) theNode = new NormAckingNode(*this, nodeId); if (NULL != theNode) { - theNode->Reset(NORM_ROBUST_FACTOR); + theNode->Reset(GetTxRobustFactor()); acking_node_tree.AttachNode(theNode); acking_node_count++; return true; @@ -990,7 +967,8 @@ bool NormSession::ServerQueueWatermarkFlush() } if (watermark_pending) { - if (flush_count < NORM_ROBUST_FACTOR) flush_count++; + if ((GetTxRobustFactor() < 0) || (flush_count < GetTxRobustFactor())) + flush_count++; QueueMessage(flush); DMSG(8, "NormSession::ServeQueueWatermarkFlush() node>%lu cmd queued ...\n", LocalNodeId()); @@ -1055,7 +1033,8 @@ void NormSession::ServerQueueFlush() flush->SetFecBlockLen(obj->GetBlockSize(blockId)); flush->SetFecSymbolId(segmentId); QueueMessage(flush); - if (flush_count < NORM_ROBUST_FACTOR) flush_count++; + if ((GetTxRobustFactor() < 0) || (flush_count < GetTxRobustFactor())) + flush_count++; DMSG(4, "NormSession::ServerQueueFlush() node>%lu, flush queued (flush_count:%u)...\n", LocalNodeId(), flush_count); } @@ -1077,7 +1056,8 @@ void NormSession::ServerQueueFlush() // if all tx object state is gone ... if (ServerQueueSquelch(next_tx_object_id)) { - if (flush_count < NORM_ROBUST_FACTOR) flush_count++; + if ((GetTxRobustFactor() < 0) || (flush_count < GetTxRobustFactor())) + flush_count++; DMSG(4, "NormSession::ServerQueueFlush() node>%lu squelch queued (flush_count:%u)...\n", LocalNodeId(), flush_count); } @@ -1290,7 +1270,7 @@ bool NormSession::QueueTxObject(NormObject* obj) if (oldest->IsRepairPending() || oldest->IsPending()) { DMSG(0, "NormSession::QueueTxObject() all held objects repair pending\n"); - //posted_tx_queue_empty = false; + posted_tx_queue_empty = false; return false; } else @@ -1493,9 +1473,11 @@ void NormSession::TxSocketRecvHandler(ProtoSocket& /*theSocket*/, } } // end NormSession::TxSocketRecvHandler() + void NormSession::RxSocketRecvHandler(ProtoSocket& /*theSocket*/, ProtoSocket::Event /*theEvent*/) { + unsigned int recvCount = 0; NormMsg msg; unsigned int msgLength = NormMsg::MAX_SIZE; while (rx_socket.RecvFrom(msg.AccessBuffer(), @@ -1511,6 +1493,14 @@ void NormSession::RxSocketRecvHandler(ProtoSocket& /*theSocket*/, { DMSG(0, "NormSession::RxSocketRecvHandler() warning: received bad message\n"); } + // If our system gets very busy reading sockets, we should occasionally + // execute any timeouts to keep protocol operation smooth + if (++recvCount >= 100) + { + break; + //session_mgr.DoSystemTimeout(); + //recvCount = 0; + } } } // end NormSession::RxSocketRecvHandler() @@ -1639,8 +1629,7 @@ void NormTrace(const struct timeval& currentTime, } case NormCmdMsg::CC: { - const NormCmdCCMsg& cc = - static_cast(msg); + const NormCmdCCMsg& cc = static_cast(msg); DMSG(0, " seq>%u ", cc.GetCCSequence()); NormHeaderExtension ext; while (cc.GetNextExtension(ext)) @@ -2584,7 +2573,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor { // Resource constrained, move on to next repair request DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu " - "Warning - server is resource contrained ...\n"); + "Warning - server is resource constrained ...\n"); } continue; } @@ -3020,7 +3009,7 @@ bool NormSession::ServerBuildRepairAdv(NormCmdRepairAdvMsg& cmd) bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) { tx_repair_pending = false; - if (repair_timer.GetRepeatCount()) + if (0 != repair_timer.GetRepeatCount()) { // NACK aggregation period has ended. (incorporate accumulated repair requests) DMSG(4, "NormSession::OnRepairTimeout() node>%lu server NACK aggregation time ended.\n", @@ -3041,7 +3030,7 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) obj->TxReset(); tx_repair_mask.Unset(objectId); if (!tx_pending_mask.Set(objectId)) - DMSG(0, "NormSession::OnRepairTimeout() rx_pending_mask.Set(%hu) error (1)\n", + DMSG(0, "NormSession::OnRepairTimeout() tx_pending_mask.Set(%hu) error (1)\n", (UINT16)objectId); } else @@ -3053,7 +3042,7 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) DMSG(6, "NormSession::OnRepairTimeout() node>%lu activated obj>%hu repairs ...\n", LocalNodeId(), (UINT16)objectId); if (!tx_pending_mask.Set(objectId)) - DMSG(0, "NormSession::OnRepairTimeout() rx_pending_mask.Set(%hu) error (2)\n", + DMSG(0, "NormSession::OnRepairTimeout() tx_pending_mask.Set(%hu) error (2)\n", (UINT16)objectId); } } @@ -3067,7 +3056,6 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) repair_timer.SetInterval(holdoffInterval); // repair holdoff interval = 1*GRTT DMSG(4, "NormSession::OnRepairTimeout() node>%lu starting server " "NACK holdoff timer (%lf sec)...\n", LocalNodeId(), holdoffInterval); - } else { @@ -3385,6 +3373,7 @@ void NormSession::SetGrttProbingMode(ProbingMode probingMode) } } // end NormSession::SetGrttProbingMode() + bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) { // 1) Temporarily kill probe_timer if CMD(CC) not yet tx'd @@ -3527,10 +3516,14 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) if (next->IsActive()) { UINT8 ccFlags = 0; - if (next->IsClr()) + if (next->IsClr()) + { ccFlags |= (UINT8)NormCC::CLR; + } else if (next->IsPlr()) + { ccFlags |= (UINT8)NormCC::PLR; + } ccFlags |= (UINT8)NormCC::RTT; UINT8 rttQuantized = NormQuantizeRtt(next->GetRtt()); if (cc_slow_start) ccFlags |= (UINT8)NormCC::START; @@ -3542,24 +3535,26 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) rttQuantized, rateQuantized); //if (!next->IsClr()) next->SetActive(false); - // Deactivate any nodes who have stopped providing feedback + // "Deactivate" any nodes who have stopped providing feedback struct timeval feedbackTime = next->GetFeedbackTime(); double feedbackAge = currentTime.tv_sec - feedbackTime.tv_sec; - if (currentTime.tv_usec > feedbackTime.tv_usec) + feedbackAge += 1.0e-06 * ((double)((currentTime.tv_usec - feedbackTime.tv_usec))); + + /*if (currentTime.tv_usec > feedbackTime.tv_usec) feedbackAge += 1.0e-06*((double)(currentTime.tv_usec - feedbackTime.tv_usec)); else - feedbackAge -= 1.0e-06*((double)(feedbackTime.tv_usec - currentTime.tv_usec)); + feedbackAge -= 1.0e-06*((double)(feedbackTime.tv_usec - currentTime.tv_usec));*/ double maxFeedbackAge = 5 * MAX(grtt_advertised, next->GetRtt()); // Safety bound to compensate for computer clock coarseness // and possible sluggish feedback from slower machines // at higher norm data rates (keeps rate from being // prematurely reduced) - if (maxFeedbackAge <(10*NORM_TICK_MIN)) - maxFeedbackAge = (10*NORM_TICK_MIN); - if (feedbackAge > maxFeedbackAge) + if (maxFeedbackAge <(10*NORM_TICK_MIN)) maxFeedbackAge = (10*NORM_TICK_MIN); + unsigned int ccSeqDelta = cc_sequence - next->GetCCSequence() - 2; + if ((feedbackAge > maxFeedbackAge) && (ccSeqDelta > 5)) { - DMSG(6, "Deactivating cc node feedbackAge:%lf sec maxAge:%lf sec\n", - feedbackAge, maxFeedbackAge); + DMSG(4, "Deactivating cc node feedbackAge:%lf sec maxAge:%lf sec ccSeqDelta:%u\n", + feedbackAge, maxFeedbackAge, ccSeqDelta); next->SetActive(false); } } diff --git a/common/normSession.h b/common/normSession.h index a2a2470..6e6d7c5 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -72,6 +72,9 @@ class NormSessionMgr void ActivateTimer(ProtoTimer& timer) {timer_mgr.ActivateTimer(timer);} ProtoTimerMgr& GetTimerMgr() const {return timer_mgr;} ProtoSocket::Notifier& GetSocketNotifier() const {return socket_notifier;} + + void DoSystemTimeout() + {timer_mgr.DoSystemTimeout();} NormController* GetController() const {return controller;} private: @@ -103,6 +106,7 @@ class NormSession static const UINT16 DEFAULT_NPARITY; static const UINT16 DEFAULT_TX_CACHE_MIN; static const UINT16 DEFAULT_TX_CACHE_MAX; + static const int DEFAULT_ROBUST_FACTOR; enum ProbingMode {PROBE_NONE, PROBE_PASSIVE, PROBE_ACTIVE}; enum AckingStatus @@ -235,11 +239,22 @@ class NormSession void ServerSetWatermark(NormObjectId objectId, NormBlockId blockId, NormSegmentId segmentId); + void ServerCancelWatermark(); bool ServerAddAckingNode(NormNodeId nodeId); void ServerRemoveAckingNode(NormNodeId nodeId); AckingStatus ServerGetAckingStatus(NormNodeId nodeId); + // robust factor + void SetTxRobustFactor(int value) + {tx_robust_factor = value;} + int GetTxRobustFactor() const + {return tx_robust_factor;} + void SetRxRobustFactor(int value) + {rx_robust_factor = value;} + int GetRxRobustFactor() const + {return rx_robust_factor;} + UINT16 ServerSegmentSize() const {return segment_size;} UINT16 ServerBlockSize() const {return ndata;} UINT16 ServerNumParity() const {return nparity;} @@ -449,6 +464,7 @@ class NormSession // Server parameters and state bool is_server; + int tx_robust_factor; UINT16 instance_id; UINT16 segment_size; UINT16 ndata; @@ -528,6 +544,7 @@ class NormSession // Receiver parameters bool is_client; + int rx_robust_factor; NormNodeTree server_tree; unsigned long remote_server_buffer_size; bool unicast_nacks; diff --git a/common/normThreadTest.cpp b/common/normThreadTest.cpp index 10508a6..a3ef652 100644 --- a/common/normThreadTest.cpp +++ b/common/normThreadTest.cpp @@ -81,7 +81,7 @@ NormThreadApp::NormThreadApp() tx_msg_length(0), tx_msg_index(0) { tx_msg_timer.SetListener(this, &NormThreadApp::OnTxTimeout); - tx_msg_timer.SetInterval(0.0015); + tx_msg_timer.SetInterval(0.00001); tx_msg_timer.SetRepeat(-1); worker_thread_dispatcher.SetPromptCallback(DoWorkerEvent, this); @@ -107,7 +107,7 @@ bool NormThreadApp::OnStartup(int argc, const char*const* argv) norm_instance = NormCreateInstance(); ASSERT(NORM_INSTANCE_INVALID != norm_instance); - SetDebugLevel(2); + SetDebugLevel(3); // Set a callback that will call NormGetNextEvent() if (!dispatcher.InstallGenericInput(NormGetDescriptor(norm_instance), DoNormEvent, this)) @@ -157,9 +157,24 @@ bool NormThreadApp::OnStartup(int argc, const char*const* argv) localId); ASSERT(NORM_SESSION_INVALID != norm_session); + if(!NormSetMulticastInterface(norm_session,"eth0")) + { + fprintf(stderr, "normTest: Unable to set multicast interface to \"eth0\"\n"); + //return false; + } + NormSetGrttEstimate(norm_session, 0.250); // 1 msec initial grtt - NormSetTransmitRate(norm_session, 10.0e+06); // in bits/second + NormSetTransmitRate(norm_session, 80.0e+06); // in bits/second + + + NormSetTransmitRateBounds(norm_session, 10.0e+06, 10.0e+06); + + NormSetCongestionControl(norm_session, true); + + NormSetTxLoss(norm_session, 2.0); + + //NormSetMessageTrace(norm_session, true); //NormSetLoopback(norm_session, true); @@ -169,8 +184,8 @@ bool NormThreadApp::OnStartup(int argc, const char*const* argv) { NormSessionId sessionId = (NormSessionId)rand(); //NormStartSender(norm_session, sessionId, 1024*1024, 1400, 64, 8); - NormStartSender(norm_session, sessionId, 1024*1024, 1400, 64, 0); - norm_tx_stream = NormStreamOpen(norm_session, 1024*1024); + NormStartSender(norm_session, sessionId, 2000000, 1400, 64, 8); + norm_tx_stream = NormStreamOpen(norm_session, 1800000); // Activate tx timer and force first message transmission ActivateTimer(tx_msg_timer); @@ -181,9 +196,8 @@ bool NormThreadApp::OnStartup(int argc, const char*const* argv) // 6) If receiver, start receiver if (receiver) { - NormStartReceiver(norm_session, 1024*1024); - worker_thread_dispatcher.StartThread(); + NormStartReceiver(norm_session, 2000000); } return true; @@ -207,31 +221,39 @@ void NormThreadApp::OnShutdown() bool NormThreadApp::OnTxTimeout(ProtoTimer& /*timer*/) { //TRACE("enter NormThreadApp::OnTxTimeout() ...\n"); - if (0 == tx_msg_length) + while (1) { - // Send a new message - unsigned int sendCount = 1; - sprintf(tx_msg_buffer, "normThreadTest says hello %u ", sendCount); - unsigned int msgLength = strlen(tx_msg_buffer); - memset(tx_msg_buffer + msgLength, 'a', 900 - msgLength); - tx_msg_length = 900; - tx_msg_index = 0; - } - - unsigned int bytesHave = tx_msg_length - tx_msg_index; - unsigned int bytesWritten = - NormStreamWrite(norm_tx_stream, tx_msg_buffer + tx_msg_index, tx_msg_length - tx_msg_index); - //TRACE("wrote %u bytes to stream ...\n", bytesWritten); - if (bytesWritten == bytesHave) - { - - tx_msg_length = 0; - } - else - { - // We filled the stream buffer - tx_msg_index += bytesWritten; - tx_msg_timer.Deactivate(); + if (0 == tx_msg_length) + { + // Send a new message + unsigned int sendCount = 1; + sprintf(tx_msg_buffer, "normThreadTest says hello %u ", sendCount); + unsigned int msgLength = strlen(tx_msg_buffer); + memset(tx_msg_buffer + msgLength, 'a', 900 - msgLength); + tx_msg_length = 900; + tx_msg_index = 0; + } + + unsigned int bytesHave = tx_msg_length - tx_msg_index; + unsigned int bytesWritten = + NormStreamWrite(norm_tx_stream, tx_msg_buffer + tx_msg_index, tx_msg_length - tx_msg_index); + //TRACE("wrote %u bytes to stream ...\n", bytesWritten); + if (bytesWritten == bytesHave) + { + + tx_msg_length = 0; + } + else + { + if (0 == bytesWritten) + { + TRACE("ZERO bytes written\n"); + } + // We filled the stream buffer + tx_msg_index += bytesWritten; + if (tx_msg_timer.IsActive()) tx_msg_timer.Deactivate(); + break; + } } return true; } // end NormThreadApp::OnTxTimeout() @@ -261,7 +283,7 @@ void NormThreadApp::OnNormEvent() TRACE("NORM_TX_QUEUE_EMPTY ...\n");*/ if (!tx_msg_timer.IsActive()) { - ActivateTimer(tx_msg_timer); + //ActivateTimer(tx_msg_timer); OnTxTimeout(tx_msg_timer); } break; @@ -269,6 +291,14 @@ void NormThreadApp::OnNormEvent() case NORM_GRTT_UPDATED: break; + case NORM_CC_ACTIVE: + TRACE("NORM_CC_ACTIVE ...\n"); + break; + + case NORM_CC_INACTIVE: + TRACE("NORM_CC_INACTIVE ...\n"); + break; + case NORM_REMOTE_SENDER_NEW: case NORM_REMOTE_SENDER_ACTIVE: break; @@ -281,7 +311,7 @@ void NormThreadApp::OnNormEvent() case NORM_RX_OBJECT_UPDATED: //TRACE("NORM_RX_OBJECT_UPDATED ...\n"); updateCount++; - if ((updateCount % 100) == 0) + if ((updateCount % 1000) == 0) TRACE("updateCount:%lu\n", updateCount); worker_thread_dispatcher.PromptThread(); break; @@ -321,7 +351,7 @@ void NormThreadApp::WorkerReadStream() if (0 != bytesRead) { readCount++; - if ((readCount % 100) == 0) + if ((readCount % 1000) == 0) TRACE("readCount:%lu\n", readCount); bytesRead = 1400; }