diff --git a/VERSION.TXT b/VERSION.TXT index c733a96..ea8223f 100644 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -1,5 +1,15 @@ NORM Version History +Version 1.2b9 +============= + - Fixed bugs with NormServerNode::retrieval_pool mgmnt + (Thanks to Charlie Davis for tracking these down!!!) + - Changed ProtoDispatcher thread stuff so threads + (e.g. the NORM API thread) are only actively + signaled and interrupted when absolutely necessary + - Added support for NORM_TX_QUEUE_VACANCY notification + (for NORM_OBJECT_STREAM only at the moment) + Version 1.2b8 ============= - Fixed bug introduced during "alignment" changes @@ -14,7 +24,7 @@ Version 1.2b7 (might significantly help CC performance - Fixed NormServerNode::FreeBuffers() bug which could result in seg fault on remote server timeout - or at app shutdown\ + or at app shutdown - Fixed bug where tx_sequence count was being incremented for all transmitted messages (The "fix" does not yet address separate increment of messages diff --git a/common/normApi.cpp b/common/normApi.cpp index 6272bfc..e200015 100644 --- a/common/normApi.cpp +++ b/common/normApi.cpp @@ -229,9 +229,9 @@ void NormInstance::Notify(NormController::Event event, NormStreamObject* stream = static_cast(object); // (TBD) implement silent_client accept differently NormObjectSize size = stream->GetSize(); - // We double the size to prevent unecessary data loss + // We double the stream buffer to prevent unecessary data loss // for our threaded API - if (!stream->Accept(2*size.LSB())) + if (!stream->Accept(size.LSB(), true)) { DMSG(0, "NormInstance::Notify() stream accept error\n"); notify_pool.Append(n); @@ -897,7 +897,7 @@ NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle, if (session) { NormObject* obj = - static_cast(session->QueueTxStream(bufferSize)); + static_cast(session->QueueTxStream(bufferSize, true)); if (obj) objectHandle = (NormObjectHandle)obj; } instance->dispatcher.ResumeThread(); diff --git a/common/normNode.cpp b/common/normNode.cpp index 6a68fc4..c793293 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -204,6 +204,7 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, Close(); return false; } + memset(retrieval_pool, 0, numData*sizeof(char*)); for (UINT16 i = 0; i < numData; i++) { char* s = new char[segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()]; @@ -323,9 +324,10 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime, if (obj && (NormObject::STREAM == obj->GetType())) { NormBlockId blockId = squelch.GetFecBlockId(); - ((NormStreamObject*)obj)->Prune(blockId); + static_cast(obj)->Prune(blockId); } - // 3) (TBD) Discard any invalidated objects + // 3) (TBD) Go ahead and discard any invalidated objects + // (although they will eventually get discarded anyway) break; } diff --git a/common/normObject.cpp b/common/normObject.cpp index ddc2fc1..c01308e 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -1210,11 +1210,15 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, #endif // SIMULATE // Zeroize the missing segment payload in prep for decoding memset(segment, 0, payloadMax+1); + server->SetRetrievalLoc(retrievalCount++, nextSegment); block->SetSegment(nextSegment, segment); } else if (!block->GetSegment(nextSegment)) { - if (!(segment = RetrieveSegment(blockId, nextSegment))) + // "tempRetrieval" is set to "true" if the retrieved segment + // comes from the "server" temp retrieval pool + bool tempRetrieval = false; + if (!(segment = RetrieveSegment(blockId, nextSegment, tempRetrieval))) { ASSERT(IsStream()); block->SetPending(nextSegment); @@ -1227,7 +1231,8 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, block->DetachSegment(server->GetRetrievalLoc(i)); return; } - server->SetRetrievalLoc(retrievalCount++, nextSegment); + if (tempRetrieval) + server->SetRetrievalLoc(retrievalCount++, nextSegment); block->SetSegment(nextSegment, segment); } } @@ -1618,7 +1623,7 @@ void NormStreamObject::StreamAdvance() } else { - DMSG(0, "NormStreamObject::StreamAdvance() node>%lu Pending segment repairs (blk>%lu) " + DMSG(0, "NormStreamObject::StreamAdvance() warning: node>%lu Pending segment repairs (blk>%lu) " "delaying stream advance ...\n", LocalNodeId(), (UINT32)block->GetId()); } } @@ -1904,7 +1909,8 @@ UINT16 NormFileObject::ReadSegment(NormBlockId blockId, } // end NormFileObject::ReadSegment() char* NormFileObject::RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId) + NormSegmentId segmentId, + bool& tempRetrieval) { if (server) { @@ -1915,6 +1921,7 @@ char* NormFileObject::RetrieveSegment(NormBlockId blockId, // zeroize remainder for proper decodes if (len < segment_size) memset(segment+len, 0, segment_size-len); + tempRetrieval = true; return segment; } else @@ -2112,14 +2119,15 @@ UINT16 NormDataObject::ReadSegment(NormBlockId blockId, return len; } // end NormDataObject::ReadSegment() -char* NormDataObject::RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId) +char* NormDataObject::RetrieveSegment(NormBlockId blockId, + NormSegmentId segmentId, + bool& tempRetrieval) { if (NULL == data_ptr) { DMSG(0, "NormDataObject::RetrieveSegment() error: NULL data_ptr\n"); return NULL; - } + } // Determine segment length from blockId::segmentId UINT16 len; if (blockId == final_block_id) @@ -2133,13 +2141,29 @@ char* NormDataObject::RetrieveSegment(NormBlockId blockId, { len = segment_size; } - if (len < segment_size) + // Determine segment offset from blockId::segmentId + NormObjectSize segmentOffset; + NormObjectSize segmentSize = NormObjectSize(segment_size); + if ((UINT32)blockId < large_block_count) + { + segmentOffset = large_block_length*(UINT32)blockId + segmentSize*segmentId; + } + else + { + segmentOffset = large_block_length*large_block_count; // (TBD) pre-calc this + UINT32 smallBlockIndex = (UINT32)blockId - large_block_count; + segmentOffset = segmentOffset + small_block_length*smallBlockIndex + + segmentSize*segmentId; + } + ASSERT(0 == segmentOffset.MSB()); + if ((len < segment_size) || (data_max < (segmentOffset.LSB() + len))) { if (server) { char* segment = server->GetRetrievalSegment(); len = ReadSegment(blockId, segmentId, segment); memset(segment+len, 0, segment_size-len); + tempRetrieval = true; return segment; } else @@ -2150,22 +2174,7 @@ char* NormDataObject::RetrieveSegment(NormBlockId blockId, } else { - // Determine segment offset from blockId::segmentId - NormObjectSize segmentOffset; - NormObjectSize segmentSize = NormObjectSize(segment_size); - if ((UINT32)blockId < large_block_count) - { - segmentOffset = large_block_length*(UINT32)blockId + segmentSize*segmentId; - } - else - { - segmentOffset = large_block_length*large_block_count; // (TBD) pre-calc this - UINT32 smallBlockIndex = (UINT32)blockId - large_block_count; - segmentOffset = segmentOffset + small_block_length*smallBlockIndex + - segmentSize*segmentId; - } - ASSERT(0 == segmentOffset.MSB()); - ASSERT(data_max >= (segmentOffset.LSB() + len)); + tempRetrieval = false; return (data_ptr + segmentOffset.LSB()); } } // end NormDataObject::RetrieveSegment() @@ -2192,6 +2201,7 @@ NormStreamObject::~NormStreamObject() } bool NormStreamObject::Open(UINT32 bufferSize, + bool doubleBuffer, const char* infoPtr, UINT16 infoLen) { @@ -2216,9 +2226,9 @@ bool NormStreamObject::Open(UINT32 bufferSize, UINT32 blockSize = segmentSize * numData; UINT32 numBlocks = bufferSize / blockSize; - // Buffering requires at least 2 blocks numBlocks = MAX(2, numBlocks); + if (doubleBuffer) numBlocks *= 2; UINT32 numSegments = numBlocks * numData; if (!block_pool.Init(numBlocks, numData)) @@ -2246,7 +2256,9 @@ bool NormStreamObject::Open(UINT32 bufferSize, read_init = true; read_index.block = read_index.segment = 0; write_index.block = write_index.segment = 0; - write_offset = read_offset = 0; + tx_offset = write_offset = read_offset = 0; + write_vacancy = true; + posted_tx_queue_vacancy = false; if (!server) { @@ -2271,9 +2283,9 @@ bool NormStreamObject::Open(UINT32 bufferSize, return true; } // end NormStreamObject::Open() -bool NormStreamObject::Accept(UINT32 bufferSize) +bool NormStreamObject::Accept(UINT32 bufferSize, bool doubleBuffer) { - if (Open(bufferSize)) + if (Open(bufferSize, doubleBuffer)) { NormObject::Accept(); return true; @@ -2287,7 +2299,7 @@ bool NormStreamObject::Accept(UINT32 bufferSize) void NormStreamObject::Close() { NormObject::Close(); - write_offset = read_offset = 0; + tx_offset = write_offset = read_offset = 0; NormBlock* b; while ((b = stream_buffer.Find(stream_buffer.RangeLo()))) { @@ -2418,8 +2430,9 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) } // end NormStreamObject::StreamUpdateStatus() -char* NormStreamObject::RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId) +char* NormStreamObject::RetrieveSegment(NormBlockId blockId, + NormSegmentId segmentId, + bool& tempRetrieval) { NormBlock* block = stream_buffer.Find(blockId); if (!block) @@ -2430,10 +2443,10 @@ char* NormStreamObject::RetrieveSegment(NormBlockId blockId, char* segment = block->GetSegment(segmentId); if (NULL == segment) DMSG(0, "NormStreamObject::RetrieveSegment() segment unavailable\n"); + tempRetrieval = false; return segment; } // end NormStreamObject::RetrieveSegment() - UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, NormSegmentId segmentId, char* buffer, @@ -2454,8 +2467,33 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, return false; } block->UnsetPending(segmentId); + char* segment = block->GetSegment(segmentId); ASSERT(segment != NULL); + + // Update tx_offset if ((segmentOffset - tx_offset) > 0) + UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); + INT32 offsetDelta = segmentOffset - tx_offset; + if (offsetDelta > 0) tx_offset = segmentOffset; + + // Only advertise vacancy if stream_buffer.RangeLo() is non-pending _and_ + // (write_offset - tx_offset) > streamBufferSize + if (!write_vacancy) + { + offsetDelta = write_offset - tx_offset; + ASSERT(offsetDelta >= 0); + if ((UINT32)offsetDelta < object_size.LSB()) + { + NormBlock* b = stream_buffer.Find(stream_buffer.RangeLo()); + if (b && !b->IsPending()) write_vacancy = true; + } + } + if (write_vacancy && !posted_tx_queue_vacancy) + { + posted_tx_queue_vacancy = true; + session.Notify(NormController::TX_QUEUE_VACANCY, NULL, this); + } + UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); ASSERT(segmentLength <= segment_size); UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); @@ -2824,6 +2862,14 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) UINT32 nBytes = 0; do { + INT32 deltaOffset = write_offset - tx_offset; + ASSERT(deltaOffset >= 0); + if (deltaOffset >= (INT32)object_size.LSB()) + { + write_vacancy = false; + DMSG(8, "NormStreamObject::Write() stream buffer full (1)\n", len, eom); + break; // (TBD) skip break if push_mode is enabled??? + } NormBlock* block = stream_buffer.Find(write_index.block); if (!block) { @@ -2831,28 +2877,32 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) { block = stream_buffer.Find(stream_buffer.RangeLo()); ASSERT(block); - if (push_mode) + if (block->IsPending()) { - NormBlockId blockId = block->GetId(); - pending_mask.Unset(blockId); - repair_mask.Unset(blockId); - NormBlock* b = FindBlock(blockId); - if (b) + write_vacancy = false; + if (push_mode) { - block_buffer.Remove(b); - session.ServerPutFreeBlock(b); - } - if (!pending_mask.IsSet()) - { - pending_mask.Set(write_index.block); - stream_next_id = write_index.block + 1; + NormBlockId blockId = block->GetId(); + pending_mask.Unset(blockId); + repair_mask.Unset(blockId); + NormBlock* b = FindBlock(blockId); + if (b) + { + block_buffer.Remove(b); + session.ServerPutFreeBlock(b); + } + if (!pending_mask.IsSet()) + { + pending_mask.Set(write_index.block); + stream_next_id = write_index.block + 1; + } } - } - else if (block->IsPending()) - { - DMSG(0, "NormStreamObject::Write() stream buffer full (1) len:%d eom:%d\n", len, eom); - break; - } + else + { + DMSG(4, "NormStreamObject::Write() stream buffer full (2) len:%d eom:%d\n", len, eom); + break; + } + } stream_buffer.Remove(block); block->EmptyToPool(segment_pool); } @@ -2869,27 +2919,31 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) { NormBlock* b = stream_buffer.Find(stream_buffer.RangeLo()); ASSERT(b != block); - if (push_mode) + if (b->IsPending()) { - NormBlockId blockId = b->GetId(); - pending_mask.Unset(blockId); - repair_mask.Unset(blockId); - NormBlock* c = FindBlock(blockId); - if (c) + write_vacancy = false; + if (push_mode) { - block_buffer.Remove(c); - session.ServerPutFreeBlock(c); - } - if (!pending_mask.IsSet()) + NormBlockId blockId = b->GetId(); + pending_mask.Unset(blockId); + repair_mask.Unset(blockId); + NormBlock* c = FindBlock(blockId); + if (c) + { + block_buffer.Remove(c); + session.ServerPutFreeBlock(c); + } + if (!pending_mask.IsSet()) + { + pending_mask.Set(write_index.block); + stream_next_id = write_index.block + 1; + } + } + else { - pending_mask.Set(write_index.block); - stream_next_id = write_index.block + 1; - } - } - else if (b->IsPending()) - { - DMSG(0, "NormStreamObject::Write() stream buffer full (2)\n"); - break; + DMSG(4, "NormStreamObject::Write() stream buffer full (3)\n"); + break; + } } stream_buffer.Remove(b); b->EmptyToPool(segment_pool); @@ -2969,6 +3023,8 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) if (0 != nBytes) session.TouchServer(); } + if ((0 != nBytes) && posted_tx_queue_vacancy) + posted_tx_queue_vacancy = false; return nBytes; } // end NormStreamObject::Write() @@ -3025,8 +3081,10 @@ UINT16 NormSimObject::ReadSegment(NormBlockId blockId, } // end NormSimObject::ReadSegment() char* NormSimObject::RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId) + NormSegmentId segmentId, + bool& tempRetrieval) { + tempRetrieval = true; return server ? server->GetRetrievalSegment() : NULL; } // end NormSimObject::RetrieveSegment() diff --git a/common/normObject.h b/common/normObject.h index c8a4a58..805e453 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -94,7 +94,8 @@ class NormObject bool* msgStart = NULL) = 0; virtual char* RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId) = 0; + NormSegmentId segmentId, + bool& tempRetrieval) = 0; NackingMode GetNackingMode() const {return nacking_mode;} void SetNackingMode(NackingMode nackingMode) @@ -303,7 +304,8 @@ class NormFileObject : public NormObject char* buffer, bool* msgStart = NULL); virtual char* RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId); + NormSegmentId segmentId, + bool& tempRetrieval); private: char path[PATH_MAX]; @@ -348,7 +350,8 @@ class NormDataObject : public NormObject bool* msgStart = NULL); virtual char* RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId); + NormSegmentId segmentId, + bool& tempRetrieval); private: NormObjectSize large_block_length; @@ -369,10 +372,11 @@ class NormStreamObject : public NormObject ~NormStreamObject(); bool Open(UINT32 bufferSize, + bool doubleBuffer = false, const char* infoPtr = NULL, UINT16 infoLen = 0); void Close(); - bool Accept(UINT32 bufferSize); + bool Accept(UINT32 bufferSize, bool doubleBuffer = false); enum FlushMode { @@ -415,7 +419,8 @@ class NormStreamObject : public NormObject bool* msgStart = NULL); virtual char* RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId); + NormSegmentId segmentId, + bool& tempRetrieval); // For receive stream, we can rewind to earliest buffered offset @@ -465,6 +470,9 @@ class NormStreamObject : public NormObject NormBlockBuffer stream_buffer; Index write_index; UINT32 write_offset; + UINT32 tx_offset; + bool write_vacancy; + bool posted_tx_queue_vacancy; bool read_init; Index read_index; UINT32 read_offset; @@ -505,7 +513,8 @@ class NormSimObject : public NormObject bool* msgStart = NULL); virtual char* RetrieveSegment(NormBlockId blockId, - NormSegmentId segmentId); + NormSegmentId segmentId, + bool& tempRetrieval); }; // end class NormSimObject #endif // SIMULATE diff --git a/common/normSession.cpp b/common/normSession.cpp index 5f62cfc..b2cb71a 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -537,6 +537,11 @@ void NormSession::Serve() else tx_pending_mask.Unset(obj->GetId()); } + else if (obj->IsStream()) + { + // Is there room write to the stream + + } } else { @@ -890,6 +895,7 @@ NormDataObject* NormSession::QueueTxData(const char* dataPtr, NormStreamObject* NormSession::QueueTxStream(UINT32 bufferSize, + bool doubleBuffer, const char* infoPtr, UINT16 infoLen) { @@ -905,7 +911,7 @@ NormStreamObject* NormSession::QueueTxStream(UINT32 bufferSize, strerror(errno)); return NULL; } - if (!stream->Open(bufferSize, infoPtr, infoLen)) + if (!stream->Open(bufferSize, doubleBuffer, infoPtr, infoLen)) { DMSG(0, "NormSession::QueueTxStream() stream open error\n"); delete stream; diff --git a/common/normSession.h b/common/normSession.h index b7c2e0e..c2684b0 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -162,6 +162,7 @@ class NormSession const char* interfaceName = NULL); void StopServer(); NormStreamObject* QueueTxStream(UINT32 bufferSize, + bool doubleBuffer = false, const char* infoPtr = NULL, UINT16 infoLen = 0); NormFileObject* QueueTxFile(const char* path, diff --git a/common/normTest.cpp b/common/normTest.cpp index 8098826..4a5df73 100644 --- a/common/normTest.cpp +++ b/common/normTest.cpp @@ -35,22 +35,25 @@ int main(int argc, char* argv[]) //NormSetMessageTrace(session, true); - NormSetTxLoss(session, 1.0); // 10% packet loss + //NormSetTxLoss(session, 1.0); // 10% packet loss - NormSetGrttEstimate(session, 0.5);//0.001); // 1 msec initial grtt + NormSetGrttEstimate(session, 0.001); // 1 msec initial grtt - NormSetTransmitRate(session, 1.0e+06); // in bits/second + NormSetTransmitRate(session, 90.0e+06); // in bits/second NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK); - // Uncomment to receive own traffic - NormSetLoopback(session, true); + // Uncomment to receive your own traffic + //NormSetLoopback(session, true); // Uncomment this line to participate as a receiver //NormStartReceiver(session, 1024*1024); + // Uncomment to enable TCP-friendly congestion control + //NormSetCongestionControl(session, true); + // Uncomment the following line to start sender - NormStartSender(session, 1024*1024, 1024, 64, 0); + NormStartSender(session, 4096*1024, 1400, 64, 0); NormAddAckingNode(session, NormGetLocalNodeId(session)); @@ -59,7 +62,7 @@ int main(int argc, char* argv[]) const char* fileName = "ferrari.jpg"; // Uncomment this line to send a stream instead of the file - stream = NormStreamOpen(session, 1024*1024); + stream = NormStreamOpen(session, 4096*1024); // NORM_FLUSH_PASSIVE automatically flushes full writes to @@ -86,27 +89,39 @@ int main(int argc, char* argv[]) switch (theEvent.type) { case NORM_TX_QUEUE_EMPTY: + //TRACE("NORM_TX_QUEUE_EMPTY ...\n"); + case NORM_TX_QUEUE_VACANCY: + //TRACE("NORM_TX_QUEUE_VACANCY ...\n"); + if (NORM_OBJECT_INVALID != stream) { - if (0 == txLen) + // We loop here to keep stream buffer full .... + bool keepWriting = true; + while (keepWriting) { - // Write a message to the "txBuffer" - memset(txBuffer, 'a', 1037); - sprintf(txBuffer+1037, "normTest says hello %d ...\n", sendCount); - txLen = strlen(txBuffer); - } - txIndex += NormStreamWrite(stream, txBuffer+txIndex, (txLen - txIndex)); - if (txIndex == txLen) - { - // Instead of "NormStreamSetFlushMode(stream, NORM_FLUSH_PASSIVE)" above - // and "NormStreamMarkEom()" here, I could have used - // "NormStreamFlush(stream, true)" here to perform explicit flushing - // and EOM marking in one fell swoop. That would be a better approach - // for apps where big stream messages need to be written with - // multiple calls to "NormStreamWrite()" - NormStreamMarkEom(stream); - txLen = txIndex = 0; - sendCount++; + if (0 == txLen) + { + // Write a message to the "txBuffer" + memset(txBuffer, 'a', 1037); + sprintf(txBuffer+1037, "normTest says hello %d ...\n", sendCount); + txLen = strlen(txBuffer); + } + unsigned int want = txLen - txIndex; + unsigned int put = NormStreamWrite(stream, txBuffer+txIndex, want); + if (put != want) keepWriting = false; + txIndex += put; + if (txIndex == txLen) + { + // Instead of "NormStreamSetFlushMode(stream, NORM_FLUSH_PASSIVE)" above + // and "NormStreamMarkEom()" here, I could have used + // "NormStreamFlush(stream, true)" here to perform explicit flushing + // and EOM marking in one fell swoop. That would be a better approach + // for apps where big stream messages need to be written with + // multiple calls to "NormStreamWrite()" + NormStreamMarkEom(stream); + txLen = txIndex = 0; + sendCount++; + } } } else diff --git a/common/normVersion.h b/common/normVersion.h index c052ccf..8f54681 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.2b8" +#define VERSION "1.2b9" #endif // _NORM_VERSION diff --git a/win32/Norm.opt b/win32/Norm.opt index fbbc4f0..c9cbe29 100644 Binary files a/win32/Norm.opt and b/win32/Norm.opt differ