BodgeMaster
/
NORM-mirror
mirror of https://github.com/USNavalResearchLaboratory/norm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

v1.2b9

pull/2/head
Jeff Weston 2019-09-11 11:45:19 -04:00
parent 5b2d493872
commit 4bd6a9acb0
10 changed files with 210 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
VERSION.TXT
View File

@ -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

6
common/normApi.cpp
View File

@ -229,9 +229,9 @@ void NormInstance::Notify(NormController::Event event,
NormStreamObject* stream = static_cast<NormStreamObject*>(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<NormObject*>(session->QueueTxStream(bufferSize));
static_cast<NormObject*>(session->QueueTxStream(bufferSize, true));
if (obj) objectHandle = (NormObjectHandle)obj;
}
instance->dispatcher.ResumeThread();

6
common/normNode.cpp
View File

@ -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<NormStreamObject*>(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;
}

186
common/normObject.cpp
View File

@ -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,8 +2119,9 @@ 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)
{
@ -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);
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;
}
}
if (!pending_mask.IsSet())
else
{
pending_mask.Set(write_index.block);
stream_next_id = write_index.block + 1;
DMSG(4, "NormStreamObject::Write() stream buffer full (2) len:%d eom:%d\n", len, eom);
break;
}
}
else if (block->IsPending())
{
DMSG(0, "NormStreamObject::Write() stream buffer full (1) len:%d eom:%d\n", len, eom);
break;
}
stream_buffer.Remove(block);
block->EmptyToPool(segment_pool);
}
@ -2869,28 +2919,32 @@ 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);
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;
}
}
if (!pending_mask.IsSet())
else
{
pending_mask.Set(write_index.block);
stream_next_id = write_index.block + 1;
DMSG(4, "NormStreamObject::Write() stream buffer full (3)\n");
break;
}
}
else if (b->IsPending())
{
DMSG(0, "NormStreamObject::Write() stream buffer full (2)\n");
break;
}
stream_buffer.Remove(b);
b->EmptyToPool(segment_pool);
block_pool.Put(b);
@ -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()

21
common/normObject.h
View File

@ -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

8
common/normSession.cpp
View File

@ -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;

1
common/normSession.h
View File

@ -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,

65
common/normTest.cpp
View File

@ -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

2
common/normVersion.h
View File

@ -32,6 +32,6 @@
#ifndef _NORM_VERSION
#define _NORM_VERSION
#define VERSION "1.2b8"
#define VERSION "1.2b9"
#endif // _NORM_VERSION

BIN
win32/Norm.opt
View File

Binary file not shown.