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

v1.1b7

pull/2/head
Jeff Weston 2019-09-11 11:34:53 -04:00
parent 4244ba41d1
commit 4a40b7ec17
8 changed files with 94 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
common/normNode.cpp
View File

@ -1143,7 +1143,6 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel,
// and queue for transmission to this server node // and queue for transmission to this server node
bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
{ {
switch(repair_timer.GetRepeatCount()) switch(repair_timer.GetRepeatCount())
{ {
case 0: // hold-off time complete case 0: // hold-off time complete
@ -1375,16 +1374,17 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned
if (currentTime.tv_usec > prev_update_time.tv_sec) if (currentTime.tv_usec > prev_update_time.tv_sec)
interval += 1.0e-06*(double)(currentTime.tv_usec - prev_update_time.tv_usec); interval += 1.0e-06*(double)(currentTime.tv_usec - prev_update_time.tv_usec);
else else
interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec); interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec);
recv_accumulator += msgSize;
double rttEstimate = rtt_confirmed ? rtt_estimate : grtt_estimate; double rttEstimate = rtt_confirmed ? rtt_estimate : grtt_estimate;
if (interval < rttEstimate)
if (interval >= rttEstimate) {
recv_accumulator += msgSize;
}
else
{ {
recv_rate = ((double)(recv_accumulator)) / interval; recv_rate = ((double)(recv_accumulator)) / interval;
prev_update_time = currentTime; prev_update_time = currentTime;
recv_accumulator = 0; recv_accumulator = msgSize;
} }
} }
else else
@ -1421,6 +1421,19 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/)
struct timeval currentTime; struct timeval currentTime;
::ProtoSystemTime(currentTime); ::ProtoSystemTime(currentTime);
UpdateRecvRate(currentTime, 0); UpdateRecvRate(currentTime, 0);
if (synchronized)
{
NormObject* objMax = rx_table.Find(max_pending_object);
if (NULL != objMax)
RepairCheck(NormObject::THRU_SEGMENT,
max_pending_object,
objMax->GetMaxPendingBlockId(),
objMax->GetMaxPendingSegmentId());
else
RepairCheck(NormObject::THRU_SEGMENT, // (TBD) thru object???
max_pending_object, 0, 0);
}
} }
if (0 == activity_timer.GetRepeatCount()) if (0 == activity_timer.GetRepeatCount())
{ {

37
common/normObject.cpp
View File

@ -312,6 +312,7 @@ bool NormObject::ActivateRepairs()
// Activate partial block (segment) repairs // Activate partial block (segment) repairs
NormBlockBuffer::Iterator iterator(block_buffer); NormBlockBuffer::Iterator iterator(block_buffer);
NormBlock* block; NormBlock* block;
while ((block = iterator.GetNextBlock())) while ((block = iterator.GetNextBlock()))
{ {
if (block->ActivateRepairs(nparity)) if (block->ActivateRepairs(nparity))
@ -944,9 +945,9 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
{ {
if (!(block = server->GetFreeBlock(transport_id, blockId))) if (!(block = server->GetFreeBlock(transport_id, blockId)))
{ {
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " //DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! no free blocks ...\n", LocalNodeId(), server->GetId(), // "Warning! no free blocks ...\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id); // (UINT16)transport_id);
return; return;
} }
block->RxInit(blockId, numData, nparity); block->RxInit(blockId, numData, nparity);
@ -958,9 +959,9 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
char* segment = server->GetFreeSegment(transport_id, blockId); char* segment = server->GetFreeSegment(transport_id, blockId);
if (!segment) if (!segment)
{ {
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " //DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! no free segments ...\n", LocalNodeId(), server->GetId(), // "Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id); // (UINT16)transport_id);
return; return;
} }
UINT16 segmentLength = data.GetPayloadDataLength(); UINT16 segmentLength = data.GetPayloadDataLength();
@ -1024,9 +1025,9 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
{ {
if (!(segment = server->GetFreeSegment(transport_id, blockId))) if (!(segment = server->GetFreeSegment(transport_id, blockId)))
{ {
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " //DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! no free segments ...\n", LocalNodeId(), server->GetId(), // "Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id); // (UINT16)transport_id);
// (TBD) Dump the block ...??? // (TBD) Dump the block ...???
return; return;
} }
@ -1122,7 +1123,7 @@ NormBlock* NormObject::StealNonPendingBlock(bool excludeBlock, NormBlockId exclu
} // end NormObject::StealNonPendingBlock() } // end NormObject::StealNonPendingBlock()
// For client resource management, steals newer block resources when // For client & server resource management, steals newer block resources when
// needing resources for ordinally older blocks. // needing resources for ordinally older blocks.
NormBlock* NormObject::StealNewestBlock(bool excludeBlock, NormBlockId excludeId) NormBlock* NormObject::StealNewestBlock(bool excludeBlock, NormBlockId excludeId)
{ {
@ -1223,8 +1224,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
{ {
if (!(block = session->ServerGetFreeBlock(transport_id, blockId))) if (!(block = session->ServerGetFreeBlock(transport_id, blockId)))
{ {
DMSG(2, "NormObject::NextServerMsg() node>%lu Warning! server resource " //DMSG(2, "NormObject::NextServerMsg() node>%lu Warning! server resource "
"constrained (no free blocks).\n", LocalNodeId()); // "constrained (no free blocks).\n", LocalNodeId());
return false; return false;
} }
// Load block with zero initialized parity segments // Load block with zero initialized parity segments
@ -1243,8 +1244,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
} }
else else
{ {
DMSG(2, "NormObject::NextServerMsg() node>%lu Warning! server resource " //DMSG(2, "NormObject::NextServerMsg() node>%lu Warning! server resource "
"constrained (no free segments).\n", LocalNodeId()); // "constrained (no free segments).\n", LocalNodeId());
session->ServerPutFreeBlock(block); session->ServerPutFreeBlock(block);
return false; return false;
} }
@ -1437,8 +1438,8 @@ NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId)
} }
else else
{ {
DMSG(2, "NormObject::ServerRecoverBlock() node>%lu Warning! server resource " //DMSG(2, "NormObject::ServerRecoverBlock() node>%lu Warning! server resource "
"constrained (no free segments).\n", LocalNodeId()); // "constrained (no free segments).\n", LocalNodeId());
session->ServerPutFreeBlock(block); session->ServerPutFreeBlock(block);
return (NormBlock*)NULL; return (NormBlock*)NULL;
} }
@ -1462,8 +1463,8 @@ NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId)
} }
else else
{ {
DMSG(2, "NormObject::ServerRecoverBlock() node>%lu Warning! server resource " //DMSG(2, "NormObject::ServerRecoverBlock() node>%lu Warning! server resource "
"constrained (no free blocks).\n", LocalNodeId()); // "constrained (no free blocks).\n", LocalNodeId());
return (NormBlock*)NULL; return (NormBlock*)NULL;
} }
} // end NormObject::ServerRecoverBlock() } // end NormObject::ServerRecoverBlock()

3
common/normObject.h
View File

@ -142,6 +142,9 @@ class NormObject
bool ActivateRepairs(); bool ActivateRepairs();
bool IsRepairSet(NormBlockId blockId) {return repair_mask.Test(blockId);} bool IsRepairSet(NormBlockId blockId) {return repair_mask.Test(blockId);}
bool AppendRepairAdv(NormCmdRepairAdvMsg& cmd); bool AppendRepairAdv(NormCmdRepairAdvMsg& cmd);
NormBlockId GetMaxPendingBlockId() const {return max_pending_block;}
NormSegmentId GetMaxPendingSegmentId() const {return max_pending_segment;}
// Used by sender for resource management scheme // Used by sender for resource management scheme
NormBlock* StealNonPendingBlock(bool excludeBlock, NormBlockId excludeId = 0); NormBlock* StealNonPendingBlock(bool excludeBlock, NormBlockId excludeId = 0);

11
common/normSegment.cpp
View File

@ -82,6 +82,7 @@ char* NormSegmentPool::Get()
{ {
if (!overrun_flag) if (!overrun_flag)
{ {
DMSG(0, "NormSegmentPool::Get() warning: operating with constrained buffering resources\n");
overruns++; overruns++;
overrun_flag = true; overrun_flag = true;
} }
@ -233,7 +234,15 @@ bool NormBlock::TxReset(UINT16 numData,
char** ptr = segment_table+numData; char** ptr = segment_table+numData;
while (numParity--) while (numParity--)
{ {
if (*ptr) memset(*ptr, 0, segmentSize); if (*ptr)
{
UINT16 payloadMax = segmentSize +
NormDataMsg::GetStreamPayloadHeaderLength();
#ifdef SIMULATE
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
#endif // SIMULATE
memset(*ptr, 0, payloadMax+1);
}
ptr++; ptr++;
} }
erasure_count = 0; erasure_count = 0;

1
common/normSegment.h
View File

@ -261,6 +261,7 @@ class NormBlockPool
} }
else if (!overrun_flag) else if (!overrun_flag)
{ {
DMSG(0, "NormBlockPool::Get() warning: operating with constrained buffering resources\n");
overruns++; overruns++;
overrun_flag = true; overrun_flag = true;
} }

64
common/normSession.cpp
View File

@ -2,6 +2,7 @@
#include <errno.h> #include <errno.h>
#include <time.h> // for gmtime() in NormTrace() #include <time.h> // for gmtime() in NormTrace()
const UINT8 NormSession::DEFAULT_TTL = 255; // bits/sec const UINT8 NormSession::DEFAULT_TTL = 255; // bits/sec
const double NormSession::DEFAULT_TRANSMIT_RATE = 64000.0; // bits/sec const double NormSession::DEFAULT_TRANSMIT_RATE = 64000.0; // bits/sec
const double NormSession::DEFAULT_GRTT_INTERVAL_MIN = 1.0; // sec const double NormSession::DEFAULT_GRTT_INTERVAL_MIN = 1.0; // sec
@ -197,7 +198,6 @@ bool NormSession::StartServer(unsigned long bufferSpace,
if (numBlocks < 2) numBlocks = 2; if (numBlocks < 2) numBlocks = 2;
unsigned long numSegments = numBlocks * numParity; unsigned long numSegments = numBlocks * numParity;
if (!block_pool.Init(numBlocks, blockSize)) if (!block_pool.Init(numBlocks, blockSize))
{ {
DMSG(0, "NormSession::StartServer() block_pool init error\n"); DMSG(0, "NormSession::StartServer() block_pool init error\n");
@ -890,8 +890,6 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
if (trace) NormTrace(currentTime, LocalNodeId(), msg, false); if (trace) NormTrace(currentTime, LocalNodeId(), msg, false);
switch (msg.GetType()) switch (msg.GetType())
{ {
case NormMsg::INFO: case NormMsg::INFO:
@ -907,8 +905,7 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
if (IsClient()) ClientHandleCommand(currentTime, (NormCmdMsg&)msg); if (IsClient()) ClientHandleCommand(currentTime, (NormCmdMsg&)msg);
break; break;
case NormMsg::NACK: case NormMsg::NACK:
DMSG(4, "NormSession::HandleReceiveMessage(NormMsg::NACK) node>%lu ...\n", DMSG(4, "NormSession::HandleReceiveMessage(NormMsg::NACK) node>%lu ...\n", LocalNodeId());
LocalNodeId());
if (IsServer() && (((NormNackMsg&)msg).GetServerId() == LocalNodeId())) if (IsServer() && (((NormNackMsg&)msg).GetServerId() == LocalNodeId()))
{ {
ServerHandleNackMessage(currentTime, (NormNackMsg&)msg); ServerHandleNackMessage(currentTime, (NormNackMsg&)msg);
@ -924,6 +921,10 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
} }
} }
if (IsClient()) ClientHandleNackMessage((NormNackMsg&)msg); if (IsClient()) ClientHandleNackMessage((NormNackMsg&)msg);
if ((3 == LocalNodeId()) && (currentTime.tv_sec > 26))
{
int x = 5;
}
break; break;
case NormMsg::ACK: case NormMsg::ACK:
if (IsServer() && (((NormAckMsg&)msg).GetServerId() == LocalNodeId())) if (IsServer() && (((NormAckMsg&)msg).GetServerId() == LocalNodeId()))
@ -1606,7 +1607,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
continue; continue;
} }
} }
} }
freshBlock = false; freshBlock = false;
numErasures = extra_parity; numErasures = extra_parity;
prevBlockId = nextBlockId; prevBlockId = nextBlockId;
@ -1628,13 +1629,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
if (1 == (txBlockIndex - nextBlockId)) if (1 == (txBlockIndex - nextBlockId))
{ {
// We're currently sending this block // We're currently sending this block
ASSERT(block->IsPending()); if (block->IsPending())
NormSegmentId firstPending = 0; {
block->GetFirstPending(firstPending); NormSegmentId firstPending = 0;
if (lastLockId <= firstPending) block->GetFirstPending(firstPending);
attemptLock = false; if (lastLockId <= firstPending)
else if (nextSegmentId < firstPending) attemptLock = false;
firstLockId = firstPending; else if (nextSegmentId < firstPending)
firstLockId = firstPending;
}
else
{
// block was just recovered
}
} }
else else
{ {
@ -1692,13 +1699,20 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
else if (1 == (txBlockIndex - nextBlockId)) else if (1 == (txBlockIndex - nextBlockId))
{ {
NormSegmentId firstPending = 0; NormSegmentId firstPending = 0;
if (!block->GetFirstPending(firstPending)) ASSERT(0); if (block->GetFirstPending(firstPending))
if (nextSegmentId > firstPending) {
if (nextSegmentId > firstPending)
object->TxUpdateBlock(block, nextSegmentId, lastSegmentId, numErasures);
else if (lastSegmentId > firstPending)
object->TxUpdateBlock(block, firstPending, lastSegmentId, numErasures);
else if (numErasures > block->ParityCount())
object->TxUpdateBlock(block, firstPending, firstPending, numErasures);
}
else
{
// This block was just recovered, so do full update
object->TxUpdateBlock(block, nextSegmentId, lastSegmentId, numErasures); object->TxUpdateBlock(block, nextSegmentId, lastSegmentId, numErasures);
else if (lastSegmentId > firstPending) }
object->TxUpdateBlock(block, firstPending, lastSegmentId, numErasures);
else if (numErasures > block->ParityCount())
object->TxUpdateBlock(block, firstPending, firstPending, numErasures);
} }
} }
} }
@ -2169,18 +2183,18 @@ bool NormSession::SendMessage(NormMsg& msg)
interval += 1.0e-06*(double)(currentTime.tv_usec - prev_update_time.tv_usec); interval += 1.0e-06*(double)(currentTime.tv_usec - prev_update_time.tv_usec);
else else
interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec); interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec);
sent_rate += 0.125 * (((double)msgSize)/interval - sent_rate); /*sent_rate += 0.125 * (((double)msgSize)/interval - sent_rate);
prev_update_time = currentTime; prev_update_time = currentTime; */
/*if (interval < grtt_advertised) if (interval < grtt_advertised)
{ {
sent_accumulator += msgSize; sent_accumulator += msgSize;
} }
else else
{ {
sent_rate = ((double)(sent_accumulator+msgSize)) / interval; sent_rate = ((double)(sent_accumulator)) / interval;
prev_update_time = currentTime; prev_update_time = currentTime;
sent_accumulator = 0; sent_accumulator = msgSize;
}*/ }
} }
else else
{ {

2
common/normVersion.h
View File

@ -32,6 +32,6 @@
#ifndef _NORM_VERSION #ifndef _NORM_VERSION
#define _NORM_VERSION #define _NORM_VERSION
#define VERSION "1.1b6" #define VERSION "1.1b7"
#endif // _NORM_VERSION #endif // _NORM_VERSION

2
unix/Makefile.common
View File

@ -13,7 +13,7 @@ NS = ../ns
INCLUDES = $(TCL_INCL_PATH) $(SYSTEM_INCLUDES) -I$(UNIX) -I$(COMMON) -I$(PROTOLIB)/common INCLUDES = $(TCL_INCL_PATH) $(SYSTEM_INCLUDES) -I$(UNIX) -I$(COMMON) -I$(PROTOLIB)/common
CFLAGS = -g -DPROTO_DEBUG -DUNIX -O -fPIC $(SYSTEM_HAVES) $(INCLUDES) CFLAGS = -DPROTO_DEBUG -DUNIX -O -fPIC $(SYSTEM_HAVES) $(INCLUDES)
LDFLAGS = $(SYSTEM_LDFLAGS) LDFLAGS = $(SYSTEM_LDFLAGS)