From a5e5d140e34a6b7dacf615e32de3919334548177 Mon Sep 17 00:00:00 2001 From: Jeff Weston Date: Wed, 11 Sep 2019 11:41:23 -0400 Subject: [PATCH] v1.2b3 --- common/normApi.cpp | 18 +++++- common/normApi.h | 2 + common/normApp.cpp | 21 ++++++- common/normMessage.h | 1 + common/normNode.cpp | 38 +++++++----- common/normNode.h | 2 +- common/normObject.cpp | 21 +++++-- common/normSession.cpp | 75 ++++++++++++------------ common/normSession.h | 3 + common/normTest.cpp | 128 ++++++++++++++++++++++++++++------------- common/normVersion.h | 2 +- 11 files changed, 205 insertions(+), 106 deletions(-) diff --git a/common/normApi.cpp b/common/normApi.cpp index b919200..6591d8d 100644 --- a/common/normApi.cpp +++ b/common/normApi.cpp @@ -1018,14 +1018,28 @@ bool NormReadStream(NormObjectHandle streamHandle, NormStreamObject* stream = static_cast((NormObject*)streamHandle); if (stream) - { result = stream->Read(buffer, numBytes); - } instance->dispatcher.ResumeThread(); } return result; } // end NormReadStream() +bool NormFindStreamMsgStart(NormObjectHandle streamHandle) +{ + bool result = false; + NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormStreamObject* stream = + static_cast((NormObject*)streamHandle); + unsigned int numBytes = 0; + if (stream) + result = stream->Read(NULL, &numBytes, true); + instance->dispatcher.ResumeThread(); + } + return result; +} // end NormFindStreamMsgStart() + NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, const char* fileName, const char* infoPtr, diff --git a/common/normApi.h b/common/normApi.h index 0984944..028039e 100644 --- a/common/normApi.h +++ b/common/normApi.h @@ -186,6 +186,8 @@ bool NormReadStream(NormObjectHandle streamHandle, char* buffer, unsigned int* numBytes); +bool NormFindStreamMsgStart(NormObjectHandle streamHandle); + // NormFileObject Functions NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, const char* fileName, diff --git a/common/normApp.cpp b/common/normApp.cpp index fbf31e7..c3ecf37 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -108,6 +108,7 @@ class NormApp : public NormController, public ProtoApp // NormSession client-only parameters unsigned long rx_buffer_size; // bytes + unsigned int rx_sock_buffer_size; NormFileList rx_file_cache; char* rx_cache_path; NormPostProcessor* post_processor; @@ -129,7 +130,7 @@ NormApp::NormApp() push_mode(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE), input_messaging(false), input_msg_length(0), input_msg_index(0), output_index(0), output_messaging(false), output_msg_length(0), output_msg_sync(false), - address(NULL), port(0), ttl(3), loopback(false), interface_name(NULL), + address(NULL), port(0), ttl(32), loopback(false), interface_name(NULL), tx_rate(64000.0), cc_enable(false), segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0), backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR), @@ -137,7 +138,8 @@ NormApp::NormApp() group_size(NormSession::DEFAULT_GSIZE_ESTIMATE), tx_buffer_size(1024*1024), tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true), - rx_buffer_size(1024*1024), rx_cache_path(NULL), unicast_nacks(false), silent_client(false), + rx_buffer_size(1024*1024), rx_sock_buffer_size(0), + rx_cache_path(NULL), unicast_nacks(false), silent_client(false), tracing(false), tx_loss(0.0), rx_loss(0.0) { @@ -215,6 +217,7 @@ const char* const NormApp::cmd_list[] = "+gsize", // Set sender's group size estimate "+txbuffer", // Size of sender's buffer "+rxbuffer", // Size receiver allocates for buffering each sender + "+rxsockbuffer", // Optional recv socket buffer size. "-unicastNacks", // unicast instead of multicast feedback messages "-silentClient", // "silent" (non-nacking) client (EMCON mode) "+processor", // receive file post processing command @@ -687,6 +690,16 @@ bool NormApp::OnCommand(const char* cmd, const char* val) return false; } } + else if (!strncmp("rxsockbuffer", cmd, len)) + { + if (1 != sscanf(val, "%u", &rx_sock_buffer_size)) + { + DMSG(0, "NormApp::OnCommand(rxsockbuffer) invalid value!\n"); + return false; + } + if (session && (rx_sock_buffer_size > 0)) + session->SetRxSocketBuffer(rx_sock_buffer_size); + } else if (!strncmp("unicastNacks", cmd, len)) { unicast_nacks = true; @@ -1186,7 +1199,7 @@ void NormApp::Notify(NormController::Event event, DMSG(0, "NormApp::Notify() detected broken stream ...\n"); output_msg_length = output_index = 0; output_msg_sync = false; - continue; + break; } } else @@ -1464,6 +1477,8 @@ bool NormApp::OnStartup(int argc, const char*const* argv) session_mgr.Destroy(); return false; } + if (rx_sock_buffer_size > 0) + session->SetRxSocketBuffer(rx_sock_buffer_size); } return true; } diff --git a/common/normMessage.h b/common/normMessage.h index 21c6146..4d0f6c5 100644 --- a/common/normMessage.h +++ b/common/normMessage.h @@ -198,6 +198,7 @@ class NormObjectId bool operator!=(const NormObjectId& id) const {return (value != id.value);} NormObjectId& operator++(int) {value++; return *this;} + NormObjectId& operator--(int) {value--; return *this;} private: UINT16 value; diff --git a/common/normNode.cpp b/common/normNode.cpp index 6896da5..99c0063 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -207,7 +207,9 @@ void NormServerNode::FreeBuffers() while ((obj = rx_table.Find(rx_table.RangeLo()))) { session.Notify(NormController::RX_OBJECT_ABORTED, this, obj); - DeleteObject(obj, 6); + DeleteObject(obj); + // We do the following to remember which objects were pending + rx_pending_mask.Set(obj->GetId()); } segment_pool.Destroy(); block_pool.Destroy(); @@ -685,7 +687,7 @@ void NormServerNode::CalculateGrttResponse(const struct timeval& currentTime, } } // end NormServerNode::CalculateGrttResponse() -void NormServerNode::DeleteObject(NormObject* obj, int which) +void NormServerNode::DeleteObject(NormObject* obj) { if (rx_table.Remove(obj)) rx_pending_mask.Unset(obj->GetId()); @@ -866,10 +868,16 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) } else { + // The hacky use of "sync_id" here keeps the debug message from + // printing too often. if (0 == sync_id) { DMSG(0, "NormServerNode::HandleObjectMessage() waiting to sync ...\n"); - sync_id = 1; + sync_id = 100; + } + else + { + sync_id--; } return; } @@ -939,13 +947,13 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) } else { - DeleteObject(obj, 7); + DeleteObject(obj); obj = NULL; } } else { - DeleteObject(obj, 1); + DeleteObject(obj); obj = NULL; } break; @@ -955,7 +963,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) { DMSG(0, "NormServerNode::HandleObjectMessage() node>%lu server>%lu " "new obj>%hu - no FTI provided!\n", LocalNodeId(), GetId(), (UINT16)objectId); - DeleteObject(obj, 2); + DeleteObject(obj); obj = NULL; } } @@ -985,7 +993,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) { // Streams never complete session.Notify(NormController::RX_OBJECT_COMPLETED, this, obj); - DeleteObject(obj, 3); + DeleteObject(obj); completion_count++; } } @@ -1058,7 +1066,7 @@ void NormServerNode::Sync(NormObjectId objectId) while ((obj = rx_table.Find(rx_table.RangeLo()))) { session.Notify(NormController::RX_OBJECT_ABORTED, this, obj); - DeleteObject(obj, 4); + DeleteObject(obj); failure_count++; } rx_pending_mask.Clear(); @@ -1070,7 +1078,7 @@ void NormServerNode::Sync(NormObjectId objectId) (obj->GetId() < objectId)) { session.Notify(NormController::RX_OBJECT_ABORTED, this, obj); - DeleteObject(obj, 5); + DeleteObject(obj); failure_count++; } unsigned long numBits = (UINT16)(objectId - firstPending) + 1; @@ -1583,15 +1591,15 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned else interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec); double rttEstimate = rtt_confirmed ? rtt_estimate : grtt_estimate; - if (interval < rttEstimate) - { - recv_accumulator += msgSize; - } - else + // We put a 0.100 sec lower bound on our rttEstimate for the recv_rate measurement + // interval because of the typical limited granularity of our system clock + rttEstimate = rttEstimate < 0.1 ? 0.1 : rttEstimate; + recv_accumulator += msgSize; + if (interval >= rttEstimate) { recv_rate = ((double)(recv_accumulator)) / interval; prev_update_time = currentTime; - recv_accumulator = msgSize; + recv_accumulator = 0; } } else diff --git a/common/normNode.h b/common/normNode.h index 553243b..cdd8678 100644 --- a/common/normNode.h +++ b/common/normNode.h @@ -281,7 +281,7 @@ class NormServerNode : public NormNode } void SetPending(NormObjectId objectId); - void DeleteObject(NormObject* obj, int which); + void DeleteObject(NormObject* obj); UINT16 SegmentSize() {return segment_size;} UINT16 BlockSize() {return ndata;} diff --git a/common/normObject.cpp b/common/normObject.cpp index cae4677..1857e18 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -2315,7 +2315,8 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, block = stream_buffer.Find(stream_buffer.RangeLo()); if (block->IsPending()) broken = true; // This loop feeds any received user data segments to the application - // (if the application doesn't want the data, it's lost) + // (if the application doesn't want the data, it's lost!) + bool dataLost = false; while (block->IsPending()) { // Notify app for stream data salvage @@ -2330,9 +2331,8 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, if (tempOffset == read_offset) { // App didn't want any data here, purge segment - DMSG(0, "NormStreamObject::WriteSegment() app didn't want data ?!, blockId:%lu\n", - (UINT32)block->GetId()); - ASSERT(0); + //ASSERT(0); + dataLost = true; char* s = block->DetachSegment(read_index.segment); segment_pool.Put(s); block->UnsetPending(read_index.segment); @@ -2345,6 +2345,9 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, break; } } + if (dataLost) + DMSG(0, "NormStreamObject::WriteSegment() broken stream data dropped!\n"); + if (block) { // if the app didn't consume the block, we must @@ -2443,7 +2446,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar { // DMSG(0, "NormStreamObject::Read() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty()); *buflen = bytesRead; - return true; + if (bytesRead > 0) + return true; + else + return findMsgStart ? false : true; } char* segment = block->Segment(read_index.segment); if (!segment) @@ -2451,7 +2457,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar //DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n", // (UINT32)read_index.block, read_index.segment); *buflen = bytesRead; - return true; + if (bytesRead > 0) + return true; + else + return findMsgStart ? false : true; } UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); diff --git a/common/normSession.cpp b/common/normSession.cpp index 094361f..563b6cb 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -1378,22 +1378,26 @@ double NormSession::CalculateRtt(const struct timeval& currentTime, void NormSession::ServerUpdateGrttEstimate(double clientRtt) { grtt_response = true; - if ((clientRtt > grtt_current_peak) || !address.IsMulticast()) + //if ((clientRtt > grtt_current_peak) || !address.IsMulticast()) + if ((clientRtt > grtt_measured) || !address.IsMulticast()) { // Immediately incorporate bigger RTT's grtt_current_peak = clientRtt; - if ((clientRtt > grtt_measured) || !address.IsMulticast()) + //if ((clientRtt > grtt_measured) || !address.IsMulticast()) { grtt_decrease_delay_count = DEFAULT_GRTT_DECREASE_DELAY; - grtt_measured = 0.25 * grtt_measured + 0.75 * clientRtt; + //grtt_measured = 0.25 * grtt_measured + 0.75 * clientRtt; + grtt_measured = 0.9 * grtt_measured + 0.1 * clientRtt; if (grtt_measured > grtt_max) grtt_measured = grtt_max; double pktInterval = ((double)(44+segment_size))/tx_rate; UINT8 grttQuantizedOld = grtt_quantized; grtt_quantized = NormQuantizeRtt(MAX(pktInterval, grtt_measured)); // Calculate grtt_advertised since quantization rounds upward grtt_advertised = NormUnquantizeRtt(grtt_quantized); + + double clrRtt = cc_node_list.Head() ? ((NormCCNode*)cc_node_list.Head())->GetRtt() : -1; if (grttQuantizedOld != grtt_quantized) - DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt: %lf sec.\n", + DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt>%lf sec\n", LocalNodeId(), grtt_advertised); } } @@ -1433,7 +1437,7 @@ void NormSession::ServerHandleCCFeedback(NormNodeId nodeId, } if (!cc_enable) return; - // Adjust ccRtt if we have state on this nodeId + // Adjust ccRtt if we already have state on this nodeId NormCCNode* node = (NormCCNode*)cc_node_list.FindNodeById(nodeId); if (node) ccRtt = node->UpdateRtt(ccRtt); @@ -1818,7 +1822,8 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor if ((OBJECT == requestLevel) || (INFO == requestLevel)) { nextObjectId++; - if (nextObjectId > lastObjectId) inRange = false; + if (nextObjectId > lastObjectId) + inRange = false; } else { @@ -1842,7 +1847,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor startTimer = true; } } - } + } // end if (freshObject) ASSERT(object); switch (requestLevel) @@ -1871,11 +1876,10 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor nextObjectId++; if (nextObjectId > lastObjectId) inRange = false; break; - case BLOCK: DMSG(8, "NormSession::ServerHandleNackMessage(BLOCK) obj>%hu blks>%lu:%lu\n", (UINT16)nextObjectId, (UINT32)nextBlockId, (UINT32)lastBlockId); - + inRange = false; // BLOCK requests are processed in one pass // (TBD) if entire object is TxReset(), continue if (object->IsStream()) { @@ -1893,7 +1897,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } else if (nextObjectId < txObjectIndex) { - attemptLock = false; // NACK arrived too late + attemptLock = false; // NACK arrived too late to be useful } } @@ -1907,21 +1911,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor { DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu LockBlocks() failure\n", LocalNodeId()); - inRange = false; if (!squelchQueued) { ServerQueueSquelch(nextObjectId); squelchQueued = true; } - continue; + break; } } else { - inRange = false; - continue; + break; // ignore late arriving NACK } - } + } // end if (object->IsStream() if (holdoff) { if (nextObjectId == txObjectIndex) @@ -1946,21 +1948,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor object->HandleBlockRequest(nextBlockId, lastBlockId); startTimer = true; } - inRange = false; break; case SEGMENT: DMSG(8, "NormSession::ServerHandleNackMessage(SEGMENT) obj>%hu blk>%lu segs>%hu:%hu\n", (UINT16)nextObjectId, (UINT32)nextBlockId, (UINT32)nextSegmentId, (UINT32)lastSegmentId); + inRange = false; // SEGMENT repairs are also handled in one pass if (nextBlockId != prevBlockId) freshBlock = true; if (freshBlock) { + freshBlock = false; // Is this entire block already repair pending? if (object->IsRepairSet(nextBlockId)) - { - inRange = false; - continue; - } + break; if (!(block = object->FindBlock(nextBlockId))) { // Is this entire block already tx pending? @@ -1974,22 +1974,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu " "recvd repair request for old stream block(%lu) ...\n", LocalNodeId(), (UINT32)nextBlockId); - inRange = false; if (!squelchQueued) { ServerQueueSquelch(nextObjectId); squelchQueued = true; } - continue; } else { // Resource constrained, move on. DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu " "Warning - server is resource contrained ...\n"); - inRange = false; - continue; } + break; } } else @@ -1997,10 +1994,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor // Entire block already tx pending, don't recover DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu " "recvd SEGMENT repair request for pending block.\n"); - continue; + break; } } - freshBlock = false; numErasures = extra_parity; prevBlockId = nextBlockId; } @@ -2052,21 +2048,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor { DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu " "LockSegments() failure\n", LocalNodeId()); - inRange = false; if (!squelchQueued) { ServerQueueSquelch(nextObjectId); squelchQueued = true; } - continue; + break; } } else { - inRange = false; - continue; + break; // ignore late arriving NACK } - } // end (object->IsStream() && (nextSegmentId < ndata)) + } // end if (object->IsStream() && (nextSegmentId < ndata)) // With a series of SEGMENT repair requests for a block, "numErasures" will // eventually total the number of missing segments in the block. @@ -2115,12 +2109,11 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor nparity, numErasures); startTimer = true; } // end if/else (holdoff) - inRange = false; break; - case INFO: nextObjectId++; - if (nextObjectId > lastObjectId) inRange = false; + if (nextObjectId > lastObjectId) + inRange = false; break; } // end switch(requestLevel) } // end while(inRange) @@ -2841,7 +2834,7 @@ void NormSession::AdjustRate(bool onResponse) } } - DMSG(6, "ServerRateTracking time>%lf rate>%lf rtt>%lf loss>%lf\n\n", theTime, tx_rate*(8.0/1000.0), ccRtt, ccLoss); + DMSG(8, "ServerRateTracking time>%lf rate>%lf rtt>%lf loss>%lf\n\n", theTime, tx_rate*(8.0/1000.0), ccRtt, ccLoss); } // end NormSession::AdjustRate() bool NormSession::OnReportTimeout(ProtoTimer& /*theTimer*/) @@ -2855,8 +2848,16 @@ bool NormSession::OnReportTimeout(ProtoTimer& /*theTimer*/) if (IsServer()) { DMSG(2, "Local status:\n"); - DMSG(2, " txRate>%9.3lf kbps grtt>%lf\n", - ((double)tx_rate)*8.0/1000.0, grtt_advertised); + DMSG(2, " txRate>%9.3lf kbps sentRate>%9.3lf grtt>%lf\n", + ((double)tx_rate)*8.0/1000.0, sent_rate*8.0/1000.0, grtt_advertised); + if (cc_enable) + { + const NormCCNode* clr = (const NormCCNode*)cc_node_list.Head(); + if (clr) + DMSG(2, " clr>%lu rate>%9.3lf rtt>%lf loss>%lf\n", clr->GetId(), + clr->GetRate()*8.0/1000.0, clr->GetRtt(), clr->GetLoss()); + } + } if (IsClient()) { diff --git a/common/normSession.h b/common/normSession.h index 88140d9..76562ff 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -118,6 +118,9 @@ class NormSession NormSessionMgr& GetSessionMgr() {return session_mgr;} + bool SetRxSocketBuffer(unsigned int bufferSize) + {return rx_socket.SetRxBufferSize(bufferSize);} + // Session parameters double TxRate() {return (tx_rate * 8.0);} // (TBD) watch timer scheduling and min/max bounds diff --git a/common/normTest.cpp b/common/normTest.cpp index 1c66c1f..0929bf6 100644 --- a/common/normTest.cpp +++ b/common/normTest.cpp @@ -16,7 +16,7 @@ int main(int argc, char* argv[]) { printf("normTest starting ...\n"); - SetDebugLevel(4); + SetDebugLevel(2); NormInstanceHandle instance = NormCreateInstance(); @@ -33,13 +33,13 @@ int main(int argc, char* argv[]) 6003, NORM_NODE_ANY); - NormSetMessageTrace(session, true); + //NormSetMessageTrace(session, true); - NormSetTxLoss(session, 10.0); // 1% packet loss + NormSetTxLoss(session, 10.0); // 10% packet loss - NormSetGrttEstimate(session, 0.1);//0.001); // 1 msec initial grtt + NormSetGrttEstimate(session, 0.2);//0.001); // 1 msec initial grtt - NormSetTransmitRate(session, 1.0e+05); // in bits/second + NormSetTransmitRate(session, 1.0e+06); // in bits/second NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK); @@ -60,9 +60,20 @@ int main(int argc, char* argv[]) // Uncomment this line to send a stream instead of the file stream = NormOpenStream(session, 1024*1024); - NormSetStreamFlushMode(stream, NORM_FLUSH_NONE); + NormSetStreamFlushMode(stream, NORM_FLUSH_PASSIVE); - int index = -1; // used to monitor reliable stream reception + + // Some variable for stream input/output + char txBuffer[8192], rxBuffer[8192]; + int txIndex = 0; + int txLen = 0; + int rxIndex = 0; + char refBuffer[1037]; + memset(refBuffer, 'a', 1037); + bool msgSync = false; + + int msgCount = 0; + int recvCount = -1; // used to monitor reliable stream reception int sendCount = 0; int sendMax = 200; NormEvent theEvent; @@ -73,17 +84,19 @@ int main(int argc, char* argv[]) case NORM_TX_QUEUE_EMPTY: if (NORM_OBJECT_INVALID != stream) { - // Write a message to the "stream" - char buffer[1024]; - sprintf(buffer, "normTest says hello %d ...\n", sendCount++); - unsigned int len = strlen(buffer); - TRACE("writing to stream ...\n"); - if (len != NormWriteStream(stream, buffer, len)) - TRACE("incomplete write:%u\n", len); - else + if (0 == txLen) { - NormFlushStream(stream); - NormSetWatermark(session, stream); + // Write a message to the "txBuffer" + memset(txBuffer, 'a', 1037); + sprintf(txBuffer+1037, "normTest says hello %d ...\n", sendCount); + txLen = strlen(txBuffer); + } + txIndex += NormWriteStream(stream, txBuffer+txIndex, (txLen - txIndex)); + if (txIndex == txLen) + { + NormMarkStreamEom(stream); + txLen = txIndex = 0; + sendCount++; } } else @@ -102,9 +115,9 @@ int main(int argc, char* argv[]) else { TRACE("QUEUED FILE ...\n"); + sendCount++; NormSetWatermark(session, txFile); } - sendCount++; } break; @@ -152,44 +165,77 @@ int main(int argc, char* argv[]) //TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n"); if (NORM_OBJECT_STREAM != NormGetObjectType(theEvent.object)) break; - char buffer[1024]; - unsigned int len = 1023; + unsigned int len; do { - len = 1023; - if (NormReadStream(theEvent.object, buffer, &len)) + if (!msgSync) + msgSync = NormFindStreamMsgStart(theEvent.object); + if (!msgSync) break; + len = 8191 - rxIndex; + if (NormReadStream(theEvent.object, rxBuffer+rxIndex, &len)) { - buffer[len] = '\0'; - if (len) + rxIndex += len; + rxBuffer[rxIndex] = '\0'; + if (rxIndex > 0) { - TRACE("normTest: recvd(%u):\n\"%s\"\n", len, buffer); - // This while() loop is cheesy test, looking for broken stream - // - char* ptr = buffer; - while ((ptr = strstr(ptr, "hello"))) - { - int value; - if (1 == sscanf(ptr, "hello %d", &value)) + char* ptr = strchr(rxBuffer, '\n'); + if (ptr) + { + // Save sub-string length + len = ptr - rxBuffer + 1; + + // Validate string + if (memcmp(rxBuffer, refBuffer, 1037)) + ptr = (char*)NULL; + else + ptr = strstr(rxBuffer, "hello"); + if (NULL != ptr) { - if (index >= 0) + int value; + if (1 == sscanf(ptr, "hello %d", &value)) { - if (1 != (value - index)) - TRACE("WARNING! possible break? value:%d index:%d\n", - value, index); + if (recvCount >= 0) + { + if (1 != (value - recvCount)) + TRACE("WARNING! possible break? value:%d recvCount:%d\n", + value, recvCount); + else + msgCount++; // successful recv + //else + // TRACE("validated recv msg len:%d\n", len); + } + recvCount = value; + } + else + { + TRACE("couldn't find index!?\n"); + ASSERT(0); + } + if ((unsigned int)rxIndex > len) + { + memmove(rxBuffer, rxBuffer+len, rxIndex - len); + rxIndex -= len; + } + else + { + rxIndex = 0; } - index = value; } else { - TRACE("couldn't find index\n"); + TRACE("invalid string!?\n"); + ASSERT(0); } - ptr += 5; } } } else { - TRACE("normTest: error reading stream\n"); + TRACE("normTest: error reading stream\n"); + TRACE("status: msgCount:%d of total:%d (%lf)\n", + msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); + msgSync = false; + rxIndex = 0; break; } } while (0 != len); @@ -209,7 +255,7 @@ int main(int argc, char* argv[]) } // end switch(theEvent.type) // Break after sending "sendMax" messages or files - if (sendCount >= sendMax) break; + //if (sendCount >= sendMax) break; } // end while (NormGetNextEvent()) diff --git a/common/normVersion.h b/common/normVersion.h index 804dcc6..2b7cf13 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.2b2" +#define VERSION "1.2b3" #endif // _NORM_VERSION