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

v1.5b3

pull/2/head
Jeff Weston 2019-09-11 12:08:58 -04:00
parent 0194e2ba9a
commit bf41edfbff
6 changed files with 75 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/normObject.h
View File

@ -508,10 +508,11 @@ class NormStreamObject : public NormObject
bool IsReadReady() const {return read_ready;} bool IsReadReady() const {return read_ready;}
bool DetermineReadReadiness() const bool DetermineReadReadiness() //const
{ {
NormBlock* block = stream_buffer.Find(read_index.block); NormBlock* block = stream_buffer.Find(read_index.block);
return ((NULL != block) && (NULL != block->GetSegment(read_index.segment))); read_ready = ((NULL != block) && (NULL != block->GetSegment(read_index.segment)));
return read_ready;
} }
bool IsReadIndex(NormBlockId blockId, NormSegmentId segmentId) const bool IsReadIndex(NormBlockId blockId, NormSegmentId segmentId) const

12
include/normSegment.h
View File

@ -56,8 +56,8 @@ class NormBlock
NormBlock(); NormBlock();
~NormBlock(); ~NormBlock();
const NormBlockId& GetId() const {return id;} const NormBlockId& GetId() const {return blk_id;}
void SetId(NormBlockId& x) {id = x;} void SetId(NormBlockId& x) {blk_id = x;}
bool Init(UINT16 totalSize); bool Init(UINT16 totalSize);
void Destroy(); void Destroy();
@ -98,7 +98,7 @@ class NormBlock
// Sender routines // Sender routines
void TxInit(NormBlockId& blockId, UINT16 ndata, UINT16 autoParity) void TxInit(NormBlockId& blockId, UINT16 ndata, UINT16 autoParity)
{ {
id = blockId; blk_id = blockId;
pending_mask.Clear(); pending_mask.Clear();
pending_mask.SetBits(0, ndata+autoParity); pending_mask.SetBits(0, ndata+autoParity);
repair_mask.Clear(); repair_mask.Clear();
@ -111,7 +111,7 @@ class NormBlock
} }
void TxRecover(NormBlockId& blockId, UINT16 ndata, UINT16 nparity) void TxRecover(NormBlockId& blockId, UINT16 ndata, UINT16 nparity)
{ {
id = blockId; blk_id = blockId;
pending_mask.Clear(); pending_mask.Clear();
repair_mask.Clear(); repair_mask.Clear();
erasure_count = 0; erasure_count = 0;
@ -149,7 +149,7 @@ class NormBlock
// Receiver routines // Receiver routines
void RxInit(NormBlockId& blockId, UINT16 ndata, UINT16 nparity) void RxInit(NormBlockId& blockId, UINT16 ndata, UINT16 nparity)
{ {
id = blockId; blk_id = blockId;
pending_mask.Clear(); pending_mask.Clear();
pending_mask.SetBits(0, ndata+nparity); pending_mask.SetBits(0, ndata+nparity);
repair_mask.Clear(); repair_mask.Clear();
@ -254,7 +254,7 @@ class NormBlock
void EmptyToPool(NormSegmentPool& segmentPool); void EmptyToPool(NormSegmentPool& segmentPool);
private: private:
NormBlockId id; NormBlockId blk_id;
UINT16 size; UINT16 size;
char** segment_table; char** segment_table;

8
src/common/normMessage.cpp
View File

@ -150,10 +150,10 @@ bool NormRepairRequest::AppendRepairItem(UINT8 fecId,
{ {
if (RANGES == form) if (RANGES == form)
PLOG(PL_TRACE, "NormRepairRequest::AppendRepairItem-Range(fecId>%d obj>%hu blk>%lu seg>%hu) ...\n", PLOG(PL_TRACE, "NormRepairRequest::AppendRepairItem-Range(fecId>%d obj>%hu blk>%lu seg>%hu) ...\n",
fecId, (UINT16)objectId, (UINT32)blockId, (UINT32)symbolId); fecId, (UINT16)objectId, (UINT32)blockId, (UINT16)symbolId);
else else
PLOG(PL_TRACE, "NormRepairRequest::AppendRepairItem(fecId>%d obj>%hu blk>%lu seg>%hu) ...\n", PLOG(PL_TRACE, "NormRepairRequest::AppendRepairItem(fecId>%d obj>%hu blk>%lu seg>%hu) ...\n",
fecId, (UINT16)objectId, (UINT32)blockId, (UINT32)symbolId); fecId, (UINT16)objectId, (UINT32)blockId, (UINT16)symbolId);
ASSERT(NormPayloadId::IsValid(fecId)); ASSERT(NormPayloadId::IsValid(fecId));
UINT16 itemLength = RepairItemLength(fecId); UINT16 itemLength = RepairItemLength(fecId);
@ -186,8 +186,8 @@ bool NormRepairRequest::AppendRepairRange(UINT8 fecId,
UINT16 endSymbolId) UINT16 endSymbolId)
{ {
PLOG(PL_TRACE, "NormRepairRequest::AppendRepairRange(%hu:%lu:%hu->%hu:%lu:%hu) ...\n", PLOG(PL_TRACE, "NormRepairRequest::AppendRepairRange(%hu:%lu:%hu->%hu:%lu:%hu) ...\n",
(UINT16)startObjectId, (UINT32)startBlockId, (UINT32)startSymbolId, (UINT16)startObjectId, (UINT32)startBlockId, (UINT16)startSymbolId,
(UINT16)endObjectId, (UINT32)endBlockId, (UINT32)endSymbolId); (UINT16)endObjectId, (UINT32)endBlockId, (UINT16)endSymbolId);
// Note a "range" is two repair items // Note a "range" is two repair items
UINT16 rangeLength = RepairRangeLength(fecId); UINT16 rangeLength = RepairRangeLength(fecId);
if (buffer_len >= (ITEM_LIST_OFFSET+length+rangeLength)) if (buffer_len >= (ITEM_LIST_OFFSET+length+rangeLength))

3
src/common/normNode.cpp
View File

@ -2180,7 +2180,7 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
break; break;
} }
if (obj) if (NULL != obj)
{ {
if (obj->IsPending(nextId != max_pending_object)) if (obj->IsPending(nextId != max_pending_object))
{ {
@ -2246,6 +2246,7 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
// The nack had no repair request content, // The nack had no repair request content,
// perhaps because of our "nacking mode" // perhaps because of our "nacking mode"
// even though there were pending objects // even though there were pending objects
// TBD - should we avoid NACK hold-off when this happens?
PLOG(PL_DEBUG, "NormSenderNode::OnRepairTimeout() node>%lu zero content nack ...\n", PLOG(PL_DEBUG, "NormSenderNode::OnRepairTimeout() node>%lu zero content nack ...\n",
LocalNodeId()); LocalNodeId());
session.ReturnMessageToPool(nack); session.ReturnMessageToPool(nack);

82
src/common/normObject.cpp
View File

@ -1076,8 +1076,8 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
// Two PLOG() statements because one doesn't work right for some reason?! // Two PLOG() statements because one doesn't work right for some reason?!
if (PL_TRACE <= GetDebugLevel()) if (PL_TRACE <= GetDebugLevel())
{ {
PLOG(PL_TRACE, "NormObject::AppendRepairRequest() node>%lu obj>%lu, blk>%lu->%lu ", PLOG(PL_TRACE, "NormObject::AppendRepairRequest() node>%lu obj>%hu, blk>%lu->%lu ",
LocalNodeId(), (UINT32)((UINT16)GetId()), (UINT32)nextId, (UINT32)lastId); LocalNodeId(), (UINT16)GetId(), (UINT32)nextId, (UINT32)lastId);
PLOG(PL_ALWAYS, "(maxPending = %lu)\n", (UINT32)max_pending_block); PLOG(PL_ALWAYS, "(maxPending = %lu)\n", (UINT32)max_pending_block);
} }
bool appendRequest = false; bool appendRequest = false;
@ -1151,7 +1151,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
default: default:
break; break;
} // end switch(nextForm) } // end switch(nextForm)
if (block) if (NULL != block)
{ {
// Note our NACK construction is limited by "max_pending_block:max_pending_segment" // Note our NACK construction is limited by "max_pending_block:max_pending_segment"
// based on most recent transmissions from sender // based on most recent transmissions from sender
@ -1286,7 +1286,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
if (!stream->StreamUpdateStatus(blockId)) if (!stream->StreamUpdateStatus(blockId))
{ {
PLOG(PL_WARN, "NormObject::HandleObjectMessage() node:%lu sender:%lu obj>%hu blk>%lu " PLOG(PL_WARN, "NormObject::HandleObjectMessage() node:%lu sender:%lu obj>%hu blk>%lu "
"broken stream ...\n", LocalNodeId(), sender->GetId(), (UINT16)transport_id, (UINT32)blockId); "broken stream ...\n", LocalNodeId(), sender->GetId(), (UINT16)transport_id, (UINT32)blockId);
sender->IncrementResyncCount(); sender->IncrementResyncCount();
while (!stream->StreamUpdateStatus(blockId)) while (!stream->StreamUpdateStatus(blockId))
{ {
@ -1304,9 +1304,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
} }
else else
{ {
stream->StreamResync(blockId); // If we try to resync to too far back, we end up chasing our tail here
// do we just sync to the current block under these circumstances
stream->StreamResync(blockId); // - pending_mask.GetSize()/2);
break; break;
} }
} }
} }
} }
@ -1491,7 +1493,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
// so it's not called unnecessarily // so it's not called unnecessarily
if (objectUpdated && notify_on_update) if (objectUpdated && notify_on_update)
{ {
if ((NULL == stream) || stream->IsReadReady()) if ((NULL == stream) || stream->DetermineReadReadiness())
{ {
notify_on_update = false; notify_on_update = false;
session.Notify(NormController::RX_OBJECT_UPDATED, sender, this); session.Notify(NormController::RX_OBJECT_UPDATED, sender, this);
@ -2818,21 +2820,22 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId)
sender->PutFreeBlock(block); sender->PutFreeBlock(block);
} }
pending_mask.Clear(); pending_mask.Clear();
pending_mask.SetBits(blockId, pending_mask.GetSize()); pending_mask.SetBits(blockId, pending_mask.GetSize());
stream_sync = true; stream_sync = true;
stream_sync_id = blockId; stream_sync_id = blockId;
stream_next_id = blockId + pending_mask.GetSize(); stream_next_id = blockId + pending_mask.GetSize();
if (read_init) if (NULL != sender)
{ {
// This is a fresh rx stream, so init the read indices if (read_init && (NormSenderNode::SYNC_CURRENT != sender->GetSyncPolicy()))
// TBD - we may wish to do this here only for {
// a "NORM_SYNC_STREAM" sync policy (beginning of current stream) // This is a fresh rx stream, so init the read indices
// This is a "fresh" stream read_init = false;
read_init = false; read_index.block = blockId;
read_index.block = blockId; read_index.segment = 0;
read_index.segment = 0; read_offset = 0;
read_offset = 0; }
} }
// Since we're doing a resync including "read_init", dump any buffered data // Since we're doing a resync including "read_init", dump any buffered data
@ -2992,7 +2995,6 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
{ {
PLOG(PL_ERROR, "NormStreamObject::WriteSegment() diff:%lu segmentOffset:%lu < read_offset:%lu \n", PLOG(PL_ERROR, "NormStreamObject::WriteSegment() diff:%lu segmentOffset:%lu < read_offset:%lu \n",
diff, segmentOffset, read_offset); diff, segmentOffset, read_offset);
//ASSERT(0);
return false; return false;
} }
@ -3102,7 +3104,6 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
if (!read_ready) if (!read_ready)
{ {
// Did this segment make our stream ready for reading // Did this segment make our stream ready for reading
if ((blockId == read_index.block) && (segmentId == read_index.segment)) if ((blockId == read_index.block) && (segmentId == read_index.segment))
{ {
read_ready = true; read_ready = true;
@ -3253,15 +3254,20 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
if (delta > session.RcvrGetMaxDelay()) if (delta > session.RcvrGetMaxDelay())
forceForward = true; forceForward = true;
} }
else
{
NormBlockId firstPending;
if (GetFirstPending(firstPending))
{
if (read_index.block < firstPending)
forceForward = true;
}
}
if (forceForward) if (forceForward)
{ {
// Force read_index forward and try again. read_index.block++;
if (++read_index.segment >= ndata) read_index.segment = 0;
{ Prune(read_index.block, false);
read_index.block++;
read_index.segment = 0;
Prune(read_index.block, false);
}
continue; continue;
} }
else else
@ -3277,7 +3283,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
if (NULL == segment) if (NULL == segment)
{ {
//DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (read_offset>%lu) (2)\n", //DMSG(0, "NormStreamObject::ReadPrivate(%lu:%hu) stream buffer empty (read_offset>%lu) (2)\n",
// (UINT32)read_index.block, read_index.segment, read_offset); // (UINT32)read_index.block, read_index.segment, read_offset);
read_ready = false; read_ready = false;
*buflen = bytesRead; *buflen = bytesRead;
@ -3302,6 +3308,15 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
if (delta > session.RcvrGetMaxDelay()) if (delta > session.RcvrGetMaxDelay())
forceForward = true; forceForward = true;
} }
else
{
NormBlockId firstPending;
if (GetFirstPending(firstPending))
{
if (read_index.block < firstPending)
forceForward = true;
}
}
if (forceForward) if (forceForward)
{ {
// Force read_index forward and try again. // Force read_index forward and try again.
@ -3336,7 +3351,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
break; break;
default: default:
// Invalid stream control code, skip this invalid segment // Invalid stream control code, skip this invalid segment
PLOG(PL_ERROR, "NormStreamObject::Read() invalid stream control message\n"); PLOG(PL_ERROR, "NormStreamObject::ReadPrivate() invalid stream control message\n");
if (++read_index.segment >= ndata) if (++read_index.segment >= ndata)
{ {
stream_buffer.Remove(block); stream_buffer.Remove(block);
@ -3363,7 +3378,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
else else
{ {
// Skip this invalid segment // Skip this invalid segment
PLOG(PL_ERROR, "NormStreamObject::Read() node>%lu obj>%hu blk>%lu seg>%hu invalid stream segment!\n", PLOG(PL_ERROR, "NormStreamObject::ReadPrivate() node>%lu obj>%hu blk>%lu seg>%hu invalid stream segment!\n",
LocalNodeId(), (UINT16)transport_id, (UINT32)read_index.block, read_index.segment, read_offset, length); LocalNodeId(), (UINT16)transport_id, (UINT32)read_index.block, read_index.segment, read_offset, length);
if (++read_index.segment >= ndata) if (++read_index.segment >= ndata)
{ {
@ -3393,8 +3408,9 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
} }
else else
{ {
PLOG(PL_WARN, "NormStreamObject::Read() node>%lu obj>%hu blk>%lu seg>%hu broken stream!\n", PLOG(PL_WARN, "NormStreamObject::ReadPrivate() node>%lu obj>%hu blk>%lu seg>%hu broken stream (read_offset:%lu segmentOffset:%lu)!\n",
(UINT32)LocalNodeId(), (UINT16)transport_id, (UINT32)read_index.block, (UINT16)read_index.segment); (UINT32)LocalNodeId(), (UINT16)transport_id, (UINT32)read_index.block, (UINT16)read_index.segment,
read_offset, segmentOffset);
read_offset = segmentOffset; read_offset = segmentOffset;
*buflen = 0; *buflen = 0;
read_ready = false; //DetermineReadReadiness(); read_ready = false; //DetermineReadReadiness();
@ -3415,7 +3431,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
} }
else else
{ {
PLOG(PL_ERROR, "NormStreamObject::Read() node>%lu obj>%hu blk>%lu seg>%hu mangled stream! index:%hu length:%hu " PLOG(PL_ERROR, "NormStreamObject::ReadPrivate() node>%lu obj>%hu blk>%lu seg>%hu mangled stream! index:%hu length:%hu "
"read_offset:%lu segmentOffset:%lu\n", "read_offset:%lu segmentOffset:%lu\n",
LocalNodeId(), (UINT16)transport_id, (UINT32)read_index.block, read_index.segment, index, length, read_offset, segmentOffset); LocalNodeId(), (UINT16)transport_id, (UINT32)read_index.block, read_index.segment, index, length, read_offset, segmentOffset);
// Reset our read_offset ... // Reset our read_offset ...
@ -3434,7 +3450,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
// make sure msg start searches don't miss the stream end ... // make sure msg start searches don't miss the stream end ...
if (0 == NormDataMsg::ReadStreamPayloadLength(segment)) if (0 == NormDataMsg::ReadStreamPayloadLength(segment))
{ {
PLOG(PL_DEBUG, "NormStreamObject::Read() stream ended by sender 1\n"); PLOG(PL_DEBUG, "NormStreamObject::ReadPrivate() stream ended by sender 1\n");
session.Notify(NormController::RX_OBJECT_COMPLETED, sender, this); session.Notify(NormController::RX_OBJECT_COMPLETED, sender, this);
stream_closing = true; stream_closing = true;
sender->DeleteObject(this); sender->DeleteObject(this);
@ -3504,7 +3520,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
if (streamEnded) if (streamEnded)
{ {
PLOG(PL_DEBUG, "NormStreamObject::Read() stream ended by sender 2\n"); PLOG(PL_DEBUG, "NormStreamObject::ReadPrivate() stream ended by sender 2\n");
session.Notify(NormController::RX_OBJECT_COMPLETED, sender, this); session.Notify(NormController::RX_OBJECT_COMPLETED, sender, this);
stream_closing = true; stream_closing = true;
sender->DeleteObject(this); sender->DeleteObject(this);

21
src/common/normSegment.cpp
View File

@ -342,7 +342,7 @@ bool NormBlock::HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
UINT16 numData, UINT16 numParity, UINT16 erasureCount) UINT16 numData, UINT16 numParity, UINT16 erasureCount)
{ {
PLOG(PL_TRACE, "NormBlock::HandleSegmentRequest() blk>%lu seg>%hu:%hu erasures:%hu\n", PLOG(PL_TRACE, "NormBlock::HandleSegmentRequest() blk>%lu seg>%hu:%hu erasures:%hu\n",
(UINT32)id, (UINT16)nextId, (UINT16)lastId, erasureCount); (UINT32)blk_id, (UINT16)nextId, (UINT16)lastId, erasureCount);
bool increasedRepair = false; bool increasedRepair = false;
if (nextId < numData) if (nextId < numData)
{ {
@ -461,13 +461,13 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
ASSERT(0); // can't happen ASSERT(0); // can't happen
break; break;
case NormRepairRequest::ITEMS: case NormRepairRequest::ITEMS:
req.AppendRepairItem(fecId, fecM, objectId, id, numData, firstId); req.AppendRepairItem(fecId, fecM, objectId, blk_id, numData, firstId);
if (2 == segmentCount) if (2 == segmentCount)
req.AppendRepairItem(fecId, fecM, objectId, id, numData, currentId); req.AppendRepairItem(fecId, fecM, objectId, blk_id, numData, currentId);
break; break;
case NormRepairRequest::RANGES: case NormRepairRequest::RANGES:
req.AppendRepairRange(fecId, fecM, objectId, id, numData, firstId, req.AppendRepairRange(fecId, fecM, objectId, blk_id, numData, firstId,
objectId, id, numData, currentId); objectId, blk_id, numData, currentId);
break; break;
case NormRepairRequest::ERASURES: case NormRepairRequest::ERASURES:
// erasure counts not used // erasure counts not used
@ -504,7 +504,7 @@ NormObjectSize NormBlock::GetBytesPending(UINT16 numData,
} while (GetNextPending(nextId)); } while (GetNextPending(nextId));
} }
// Correct for final_segment_size, if applicable // Correct for final_segment_size, if applicable
if ((id == finalBlockId) && IsPending(numData - 1)) if ((blk_id == finalBlockId) && IsPending(numData - 1))
{ {
pendingBytes -= NormObjectSize(segmentSize); pendingBytes -= NormObjectSize(segmentSize);
pendingBytes += NormObjectSize(finalSegmentSize); pendingBytes += NormObjectSize(finalSegmentSize);
@ -594,13 +594,14 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
ASSERT(0); ASSERT(0);
break; break;
case NormRepairRequest::ITEMS: case NormRepairRequest::ITEMS:
req.AppendRepairItem(fecId, fecM, objectId, id, numData, firstId); // (TBD) error check req.AppendRepairItem(fecId, fecM, objectId, blk_id, numData, firstId); // (TBD) error check
if (2 == segmentCount) if (2 == segmentCount)
req.AppendRepairItem(fecId, fecM, objectId, id, numData, currentId); // (TBD) error check req.AppendRepairItem(fecId, fecM, objectId, blk_id, numData, currentId); // (TBD) error check
break; break;
case NormRepairRequest::RANGES: case NormRepairRequest::RANGES:
req.AppendRepairRange(fecId, fecM, objectId, id, numData, firstId, // (TBD) error check req.AppendRepairRange(fecId, fecM,
objectId, id, numData, currentId); // (TBD) error check objectId, blk_id, numData, firstId, // (TBD) error check
objectId, blk_id, numData, currentId); // (TBD) error check
break; break;
case NormRepairRequest::ERASURES: case NormRepairRequest::ERASURES:
// erasure counts not used // erasure counts not used