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

v1.1b6

pull/2/head
Jeff Weston 2019-09-11 11:34:04 -04:00
parent 25676f4b00
commit 4244ba41d1
11 changed files with 393 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
common/normApp.cpp
View File

@ -928,7 +928,7 @@ void NormApp::Notify(NormController::Event event,
if (silent_client)
size = NormObjectSize((UINT32)rx_buffer_size);
else
size = object->Size();
size = object->GetSize();
if (((NormStreamObject*)object)->Accept(size.LSB()))
{
@ -993,7 +993,7 @@ void NormApp::Notify(NormController::Event event,
strncpy(fileName, rx_cache_path, PATH_MAX);
UINT16 pathLen = strlen(rx_cache_path);
pathLen = MIN(pathLen, PATH_MAX);
UINT16 len = object->InfoLength();
UINT16 len = object->GetInfoLength();
len = MIN(len, (PATH_MAX - pathLen));
strncat(fileName, object->GetInfo(), len);
// Convert '/' in file info to directory delimiters

62
common/normNode.cpp
View File

@ -262,21 +262,24 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
{
is_clr = is_plr = false;
}
if (is_clr || is_plr)
double maxBackoff;
if (is_clr || is_plr || !session->Address().IsMulticast())
{
// Respond immediately
maxBackoff = 0.0;
if (cc_timer.IsActive()) cc_timer.Deactivate();
cc_timer.ResetRepeat();
OnCCTimeout(cc_timer);
}
else if (!cc_timer.IsActive())
else
{
if (cc_timer.IsActive()) break;
double backoffFactor = backoff_factor;
backoffFactor = MAX(backoffFactor, 4.0);
double maxBackoff = grtt_estimate*backoffFactor;
// (TBD) don't backoff for unicast sessions
maxBackoff = grtt_estimate*backoffFactor;
}
double backoffTime = (maxBackoff > 0.0) ?
ExponentialRand(maxBackoff, gsize_estimate) : 0.0;
ExponentialRand(maxBackoff, gsize_estimate) :
0.0;
// Bias backoff timeout based on our rate
double r;
if (slow_start)
@ -294,17 +297,16 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
r = MAX(r, 0.5);
r = (r - 0.5) / 0.4;
}
//TRACE("NormServerNode::HandleCommand(CC) node>%lu bias:%lf recv_rate:%lf send_rate:%lf "
//DMSG(0, "NormServerNode::HandleCommand(CC) node>%lu bias:%lf recv_rate:%lf send_rate:%lf "
// "grtt:%lf gsize:%lf\n",
// LocalNodeId(), r, recv_rate*(8.0/1000.0), send_rate*(8.0/1000.0),
//
backoffTime = 0.25 * r * maxBackoff + 0.75 * backoffTime;
cc_timer.SetInterval(backoffTime);
DMSG(4, "NormServerNode::HandleCommand() node>%lu begin CC back-off: %lf sec)...\n",
DMSG(6, "NormServerNode::HandleCommand() node>%lu begin CC back-off: %lf sec)...\n",
LocalNodeId(), backoffTime);
session->ActivateTimer(cc_timer);
}
} // end if (CC_RATE == ext.GetType())
} // end while (GetNextExtension())
break;
@ -598,7 +600,7 @@ void NormServerNode::DeleteObject(NormObject* obj)
// (TBD) Notify app of object's closing/demise?
obj->Close();
rx_table.Remove(obj);
rx_pending_mask.Unset(obj->Id());
rx_pending_mask.Unset(obj->GetId());
delete obj;
} // end NormServerNode::DeleteObject()
@ -615,13 +617,13 @@ NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId block
NormObject* obj;
while ((obj = iterator.GetNextObject()))
{
if (obj->Id() > objectId)
if (obj->GetId() > objectId)
{
break;
}
else
{
if (obj->Id() < objectId)
if (obj->GetId() < objectId)
b = obj->StealOldestBlock(false);
else
b = obj->StealOldestBlock(true, blockId);
@ -640,13 +642,13 @@ NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId block
NormObject* obj;
while ((obj = iterator.GetPrevObject()))
{
if (obj->Id() < objectId)
if (obj->GetId() < objectId)
{
break;
}
else
{
if (obj->Id() > objectId)
if (obj->GetId() > objectId)
b = obj->StealNewestBlock(false);
else
b = obj->StealNewestBlock(true, blockId);
@ -930,7 +932,7 @@ void NormServerNode::Sync(NormObjectId objectId)
{
NormObject* obj;
while ((obj = rx_table.Find(rx_table.RangeLo())) &&
(obj->Id() < objectId))
(obj->GetId() < objectId))
{
DeleteObject(obj);
failure_count++;
@ -1367,28 +1369,30 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned short msgSize)
{
double interval;
if (prev_update_time.tv_sec || prev_update_time.tv_usec)
{
interval = (double)(currentTime.tv_sec - prev_update_time.tv_sec);
double interval = (double)(currentTime.tv_sec - 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);
else
interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec);
}
else
{
recv_rate = ((double)msgSize) / grtt_estimate;
interval = 0.0;
prev_update_time = currentTime;
}
if (interval < grtt_estimate)
{
recv_accumulator += msgSize;
double rttEstimate = rtt_confirmed ? rtt_estimate : grtt_estimate;
if (interval >= rttEstimate)
{
recv_rate = ((double)(recv_accumulator)) / interval;
prev_update_time = currentTime;
recv_accumulator = 0;
}
}
else
{
recv_rate = ((double)(recv_accumulator+msgSize)) / interval;
if (send_rate > 0.0)
recv_rate = send_rate;
else
recv_rate = ((double)msgSize) / grtt_estimate;
prev_update_time = currentTime;
recv_accumulator = 0;
}
@ -1497,7 +1501,7 @@ bool NormServerNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
rtt_estimate, ccLoss);
ext.SetCCRate(NormQuantizeRate(ccRate));
}
//TRACE("NormServerNode::OnCCTimeout() node>%lu sending ACK rate:%lf kbps (rtt:%lf loss:%lf s:%lf recvRate:%lf) slow_start:%d\n",
//DMSG(0, "NormServerNode::OnCCTimeout() node>%lu sending ACK rate:%lf kbps (rtt:%lf loss:%lf s:%lf recvRate:%lf) slow_start:%d\n",
// LocalNodeId(), NormUnquantizeRate(ext.GetCCRate()) * (8.0/1000.0),
// rtt_estimate, ccLoss, nominal_packet_size, recv_rate*(8.0/1000.), slow_start);
ext.SetCCSequence(cc_sequence);

159
common/normObject.cpp
View File

@ -6,9 +6,9 @@
NormObject::NormObject(NormObject::Type theType,
class NormSession* theSession,
class NormServerNode* theServer,
const NormObjectId& objectId)
const NormObjectId& transportId)
: type(theType), session(theSession), server(theServer),
id(objectId), segment_size(0), pending_info(false), repair_info(false),
transport_id(transportId), segment_size(0), pending_info(false), repair_info(false),
current_block_id(0), next_segment_id(0),
max_pending_block(0), max_pending_segment(0),
info(NULL), info_len(0), accepted(false)
@ -35,8 +35,6 @@ NormNodeId NormObject::LocalNodeId() const
return session->LocalNodeId();
}
bool NormObject::Open(const NormObjectSize& objectSize,
const char* infoPtr,
UINT16 infoLen)
@ -248,7 +246,7 @@ bool NormObject::TxReset(NormBlockId firstBlock)
NormBlock* block;
while ((block = iterator.GetNextBlock()))
{
NormBlockId blockId = block->Id();
NormBlockId blockId = block->GetId();
if (blockId >= firstBlock)
{
increasedRepair |= block->TxReset(GetBlockSize(blockId),
@ -274,7 +272,7 @@ bool NormObject::TxResetBlocks(NormBlockId nextId, NormBlockId lastId)
}
NormBlock* block = block_buffer.Find(nextId);
if (block)
increasedRepair |= block->TxReset(GetBlockSize(block->Id()), nparity, autoParity, segment_size);
increasedRepair |= block->TxReset(GetBlockSize(block->GetId()), nparity, autoParity, segment_size);
nextId++;
}
return increasedRepair;
@ -297,7 +295,7 @@ bool NormObject::ActivateRepairs()
NormBlockId lastId;
ASSERT(GetLastRepair(lastId));
DMSG(6, "NormObject::ActivateRepairs() node>%lu obj>%hu activating blk>%lu->%lu repairs\n",
LocalNodeId(), (UINT16)id, (UINT32)nextId, (UINT32)lastId);
LocalNodeId(), (UINT16)transport_id, (UINT32)nextId, (UINT32)lastId);
repairsActivated = true;
UINT16 autoParity = session->ServerAutoParity();
do
@ -320,10 +318,10 @@ bool NormObject::ActivateRepairs()
{
repairsActivated = true;
DMSG(6, "NormObject::ActivateRepairs() node>%lu obj>%hu activated blk>%lu segment repairs ...\n",
LocalNodeId(), (UINT16)id, (UINT32)block->Id());
LocalNodeId(), (UINT16)transport_id, (UINT32)block->GetId());
// (TBD) This check can be eventually eliminated if everything else is done right
if (!pending_mask.Set(block->Id()))
DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error!\n", (UINT32)block->Id());
if (!pending_mask.Set(block->GetId()))
DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error!\n", (UINT32)block->GetId());
}
}
return repairsActivated;
@ -408,13 +406,13 @@ bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd)
ASSERT(0); // can't happen
break;
case NormRepairRequest::ITEMS:
req.AppendRepairItem(id, firstId, ndata, 0);
req.AppendRepairItem(transport_id, firstId, GetBlockSize(firstId), 0);
if (2 == blockCount)
req.AppendRepairItem(id, currentId, ndata, 0);
req.AppendRepairItem(transport_id, currentId, GetBlockSize(currentId), 0);
break;
case NormRepairRequest::RANGES:
req.AppendRepairRange(id, firstId, ndata, 0,
id, currentId, ndata, 0);
req.AppendRepairRange(transport_id, firstId, GetBlockSize(firstId), 0,
transport_id, currentId, GetBlockSize(currentId), 0);
break;
case NormRepairRequest::ERASURES:
// erasure counts not used
@ -432,7 +430,7 @@ bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd)
cmd.PackRepairRequest(req); // (TBD) error check
prevForm = NormRepairRequest::INVALID;
}
block->AppendRepairAdv(cmd, id, repair_info, ndata, segment_size); // (TBD) error check
block->AppendRepairAdv(cmd, transport_id, repair_info, GetBlockSize(currentId), segment_size); // (TBD) error check
}
}
} // end while(nextId < endId)
@ -736,7 +734,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
NormBlockId lastId;
ASSERT(GetLastPending(lastId));
DMSG(6, "NormObject::AppendRepairRequest() node>%lu obj>%hu, blk>%lu->%lu (maxPending:%lu)\n",
LocalNodeId(), (UINT16)id,
LocalNodeId(), (UINT16)transport_id,
(UINT32)nextId, (UINT32)lastId, (UINT32)max_pending_block);
bool appendRequest = false;
NormBlock* block = iterating ? block_buffer.Find(nextId) : NULL;
@ -783,18 +781,18 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
switch (nextForm)
{
case NormRepairRequest::ITEMS:
req.AppendRepairItem(id, prevId, GetBlockSize(prevId), 0); // (TBD) error check
req.AppendRepairItem(transport_id, prevId, GetBlockSize(prevId), 0); // (TBD) error check
if (2 == consecutiveCount)
{
prevId++;
req.AppendRepairItem(id, prevId, GetBlockSize(prevId), 0); // (TBD) error check
req.AppendRepairItem(transport_id, prevId, GetBlockSize(prevId), 0); // (TBD) error check
}
break;
case NormRepairRequest::RANGES:
{
NormBlockId lastId = prevId+consecutiveCount-1;
req.AppendRepairRange(id, prevId, GetBlockSize(prevId), 0,
id, lastId, GetBlockSize(lastId), 0); // (TBD) error check
req.AppendRepairRange(transport_id, prevId, GetBlockSize(prevId), 0,
transport_id, lastId, GetBlockSize(lastId), 0); // (TBD) error check
break;
}
default:
@ -820,16 +818,16 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
nack.PackRepairRequest(req); // (TBD) error check
if (flush || (nextId != max_pending_block))
{
block->AppendRepairRequest(nack, numData, nparity, id,
block->AppendRepairRequest(nack, numData, nparity, transport_id,
pending_info, segment_size); // (TBD) error check
}
else
{
if (max_pending_segment < numData)
block->AppendRepairRequest(nack, max_pending_segment, 0, id,
block->AppendRepairRequest(nack, max_pending_segment, 0, transport_id,
pending_info, segment_size); // (TBD) error check
else
block->AppendRepairRequest(nack, numData, nparity, id,
block->AppendRepairRequest(nack, numData, nparity, transport_id,
pending_info, segment_size); // (TBD) error check
}
}
@ -862,7 +860,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
req.SetForm(NormRepairRequest::ITEMS);
req.ResetFlags();
req.SetFlag(NormRepairRequest::INFO);
req.AppendRepairItem(id, 0, 0, 0); // (TBD) error check
req.AppendRepairItem(transport_id, 0, 0, 0); // (TBD) error check
} // end if/else (iterating)
if (NormRepairRequest::INVALID != prevForm)
nack.PackRepairRequest(req); // (TBD) error check
@ -885,7 +883,8 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
{
info_len = segment_size;
DMSG(0, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! info too long.\n", LocalNodeId(), server->GetId(), (UINT16)id);
"Warning! info too long.\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id);
}
memcpy(info, infoMsg.GetInfo(), info_len);
pending_info = false;
@ -896,7 +895,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
// (TBD) Verify info hasn't changed?
DMSG(6, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"received duplicate info ...\n", LocalNodeId(),
server->GetId(), (UINT16)id);
server->GetId(), (UINT16)transport_id);
}
}
else // NORM_MSG_DATA
@ -909,7 +908,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
if (!stream->StreamUpdateStatus(blockId))
{
DMSG(4, "NormObject::HandleObjectMessage() node:%lu server:%lu obj>%hu blk>%lu "
"broken stream ...\n", LocalNodeId(), server->GetId(), (UINT16)id, (UINT32)blockId);
"broken stream ...\n", LocalNodeId(), server->GetId(), (UINT16)transport_id, (UINT32)blockId);
//ASSERT(0);
// ??? Ignore this new packet and try to fix stream ???
@ -943,11 +942,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
NormBlock* block = block_buffer.Find(blockId);
if (!block)
{
if (!(block = server->GetFreeBlock(id, blockId)))
if (!(block = server->GetFreeBlock(transport_id, blockId)))
{
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! no free blocks ...\n", LocalNodeId(), server->GetId(),
(UINT16)id);
(UINT16)transport_id);
return;
}
block->RxInit(blockId, numData, nparity);
@ -956,12 +955,12 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
if (block->IsPending(segmentId))
{
// 1) Cache segment in block buffer in case its needed for decoding
char* segment = server->GetFreeSegment(id, blockId);
char* segment = server->GetFreeSegment(transport_id, blockId);
if (!segment)
{
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
(UINT16)id);
(UINT16)transport_id);
return;
}
UINT16 segmentLength = data.GetPayloadDataLength();
@ -969,7 +968,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
{
DMSG(0, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Error! segment too large ...\n", LocalNodeId(), server->GetId(),
(UINT16)id);
(UINT16)transport_id);
server->PutFreeSegment(segment);
return;
}
@ -1011,7 +1010,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
// and write any decoded data segments to object
DMSG(8, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu blk>%lu "
"completed block ...\n", LocalNodeId(), server->GetId(),
(UINT16)id, (UINT32)block->Id());
(UINT16)transport_id, (UINT32)block->GetId());
UINT16 erasureCount = 0;
UINT16 nextErasure = 0;
if (block->GetFirstPending(nextErasure))
@ -1023,11 +1022,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
server->SetErasureLoc(erasureCount++, nextErasure);
if (nextErasure < numData)
{
if (!(segment = server->GetFreeSegment(id, blockId)))
if (!(segment = server->GetFreeSegment(transport_id, blockId)))
{
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
(UINT16)id);
(UINT16)transport_id);
// (TBD) Dump the block ...???
return;
}
@ -1076,14 +1075,14 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
{
DMSG(6, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"received duplicate segment ...\n", LocalNodeId(),
server->GetId(), (UINT16)id);
server->GetId(), (UINT16)transport_id);
}
}
else
{
DMSG(6, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"received duplicate block message ...\n", LocalNodeId(),
server->GetId(), (UINT16)id);
server->GetId(), (UINT16)transport_id);
} // end if/else pending_mask.Test(blockId)
} // end if/else (NORM_MSG_INFO)
} // end NormObject::HandleObjectMessage()
@ -1104,7 +1103,7 @@ NormBlock* NormObject::StealNonPendingBlock(bool excludeBlock, NormBlockId exclu
NormBlock* block;
while ((block = iterator.GetNextBlock()))
{
NormBlockId bid = block->Id();
NormBlockId bid = block->GetId();
if (block->IsTransmitPending() ||
pending_mask.Test(bid) ||
repair_mask.Test(bid) ||
@ -1134,7 +1133,7 @@ NormBlock* NormObject::StealNewestBlock(bool excludeBlock, NormBlockId excludeId
else
{
NormBlock* block = block_buffer.Find(block_buffer.RangeHi());
if (excludeBlock && (excludeId == block->Id()))
if (excludeBlock && (excludeId == block->GetId()))
{
return NULL;
}
@ -1158,7 +1157,7 @@ NormBlock* NormObject::StealOldestBlock(bool excludeBlock, NormBlockId excludeId
else
{
NormBlock* block = block_buffer.Find(block_buffer.RangeLo());
if (excludeBlock && (excludeId == block->Id()))
if (excludeBlock && (excludeId == block->GetId()))
{
return NULL;
}
@ -1195,7 +1194,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
}
if (info) msg->SetFlag(NormObjectMsg::FLAG_INFO);
msg->SetFecId(129);
msg->SetObjectId(id);
msg->SetObjectId(transport_id);
// We currently always apply the FTI extension
NormFtiExtension fti;
@ -1222,7 +1221,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
NormBlock* block = block_buffer.Find(blockId);
if (!block)
{
if (!(block = session->ServerGetFreeBlock(id, blockId)))
if (!(block = session->ServerGetFreeBlock(transport_id, blockId)))
{
DMSG(2, "NormObject::NextServerMsg() node>%lu Warning! server resource "
"constrained (no free blocks).\n", LocalNodeId());
@ -1232,7 +1231,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
UINT16 totalBlockLen = numData + nparity;
for (UINT16 i = numData; i < totalBlockLen; i++)
{
char* s = session->ServerGetFreeSegment(id, blockId);
char* s = session->ServerGetFreeSegment(transport_id, blockId);
if (s)
{
UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength();
@ -1378,7 +1377,7 @@ void NormStreamObject::StreamAdvance()
else
{
DMSG(4, "NormStreamObject::StreamAdvance() node>%lu Pending segment repairs (blk>%lu) "
"delaying stream advance ...\n", LocalNodeId(), (UINT32)block->Id());
"delaying stream advance ...\n", LocalNodeId(), (UINT32)block->GetId());
}
}
}
@ -1391,10 +1390,10 @@ void NormStreamObject::StreamAdvance()
bool NormObject::CalculateBlockParity(NormBlock* block)
{
char buffer[NormMsg::MAX_SIZE];
UINT16 numData = GetBlockSize(block->Id());
UINT16 numData = GetBlockSize(block->GetId());
for (UINT16 i = 0; i < numData; i++)
{
UINT16 payloadLength = ReadSegment(block->Id(), i, buffer);
UINT16 payloadLength = ReadSegment(block->GetId(), i, buffer);
if (0 != payloadLength)
{
UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength();
@ -1416,7 +1415,7 @@ bool NormObject::CalculateBlockParity(NormBlock* block)
NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId)
{
NormBlock* block = session->ServerGetFreeBlock(id, blockId);
NormBlock* block = session->ServerGetFreeBlock(transport_id, blockId);
if (block)
{
UINT16 numData = GetBlockSize(blockId);
@ -1426,7 +1425,7 @@ NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId)
UINT16 totalBlockLen = numData + nparity;
for (UINT16 i = numData; i < totalBlockLen; i++)
{
char* s = session->ServerGetFreeSegment(id, blockId);
char* s = session->ServerGetFreeSegment(transport_id, blockId);
if (s)
{
UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength();
@ -1952,7 +1951,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
while (block->IsPending())
{
// Notify app for stream data salvage
read_index.block = block->Id();
read_index.block = block->GetId();
block->GetFirstPending(read_index.segment);
NormBlock* tempBlock = block;
UINT32 tempOffset = read_offset;
@ -1995,9 +1994,9 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
if (broken)
{
DMSG(4, "NormStreamObject::WriteSegment() node>%lu obj>%hu blk>%lu seg>%hu broken stream ...\n",
LocalNodeId(), (UINT16)id, (UINT32)blockId, (UINT16)segmentId);
LocalNodeId(), (UINT16)transport_id, (UINT32)blockId, (UINT16)segmentId);
NormBlock* first = stream_buffer.Find(stream_buffer.RangeLo());
read_index.block = first->Id();
read_index.block = first->GetId();
read_index.segment = 0;
}
}
@ -2029,10 +2028,10 @@ void NormStreamObject::Prune(NormBlockId blockId)
NormBlock* block;
while ((block = block_buffer.Find(block_buffer.RangeLo())))
{
if (block->Id() < blockId)
if (block->GetId() < blockId)
{
resync = true;
pending_mask.Unset(block->Id());
pending_mask.Unset(block->GetId());
block_buffer.Remove(block);
server->PutFreeBlock(block);
}
@ -2183,7 +2182,7 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len,
ASSERT(block);
if (push)
{
NormBlockId blockId = block->Id();
NormBlockId blockId = block->GetId();
pending_mask.Unset(blockId);
repair_mask.Unset(blockId);
NormBlock* b = FindBlock(blockId);
@ -2215,7 +2214,7 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len,
ASSERT(b != block);
if (push)
{
NormBlockId blockId = block->Id();
NormBlockId blockId = block->GetId();
pending_mask.Unset(blockId);
repair_mask.Unset(blockId);
NormBlock* c = FindBlock(blockId);
@ -2404,7 +2403,7 @@ NormObject* NormObjectTable::Find(const NormObjectId& objectId) const
{
if ((objectId < range_lo) || (objectId > range_hi)) return (NormObject*)NULL;
NormObject* theObject = table[((UINT16)objectId) & hash_mask];
while (theObject && (objectId != theObject->Id())) theObject = theObject->next;
while (theObject && (objectId != theObject->GetId())) theObject = theObject->next;
return theObject;
}
else
@ -2445,7 +2444,7 @@ bool NormObjectTable::CanInsert(NormObjectId objectId) const
bool NormObjectTable::Insert(NormObject* theObject)
{
const NormObjectId& objectId = theObject->Id();
const NormObjectId& objectId = theObject->GetId();
if (!range)
{
range_lo = range_hi = objectId;
@ -2468,7 +2467,7 @@ bool NormObjectTable::Insert(NormObject* theObject)
UINT16 index = ((UINT16)objectId) & hash_mask;
NormObject* prev = NULL;
NormObject* entry = table[index];
while (entry && (entry->Id() < objectId))
while (entry && (entry->GetId() < objectId))
{
prev = entry;
entry = entry->next;
@ -2477,24 +2476,24 @@ bool NormObjectTable::Insert(NormObject* theObject)
prev->next = theObject;
else
table[index] = theObject;
ASSERT((entry ? (objectId != entry->Id()) : true));
ASSERT((entry ? (objectId != entry->GetId()) : true));
theObject->next = entry;
count++;
size = size + theObject->Size();
size = size + theObject->GetSize();
return true;
} // end NormObjectTable::Insert()
bool NormObjectTable::Remove(const NormObject* theObject)
{
ASSERT(theObject);
const NormObjectId& objectId = theObject->Id();
const NormObjectId& objectId = theObject->GetId();
if (range)
{
if ((objectId < range_lo) || (objectId > range_hi)) return false;
UINT16 index = ((UINT16)objectId) & hash_mask;
NormObject* prev = NULL;
NormObject* entry = table[index];
while (entry && (entry->Id() != objectId))
while (entry && (entry->GetId() != objectId))
{
prev = entry;
entry = entry->next;
@ -2525,10 +2524,10 @@ bool NormObjectTable::Remove(const NormObject* theObject)
if ((entry = table[i]))
{
NormObjectId id = (UINT16)index + offset;
while(entry && (entry->Id() != id))
while(entry && (entry->GetId() != id))
{
if ((entry->Id() > objectId) &&
(entry->Id() < nextId)) nextId = entry->Id();
if ((entry->GetId() > objectId) &&
(entry->GetId() < nextId)) nextId = entry->GetId();
entry = entry->next;
}
@ -2536,7 +2535,7 @@ bool NormObjectTable::Remove(const NormObject* theObject)
}
} while (i != endex);
if (entry)
range_lo = entry->Id();
range_lo = entry->GetId();
else
range_lo = nextId;
range = range_hi - range_lo + 1;
@ -2561,18 +2560,18 @@ bool NormObjectTable::Remove(const NormObject* theObject)
if ((entry = table[i]))
{
NormObjectId id = (UINT16)index - offset;
//printf("Looking for id:%lu at index:%lu\n", (UINT16)id, i);
while(entry && (entry->Id() != id))
//printf("Looking for id:%lu at index:%lu\n", (UINT16)transport_id, i);
while(entry && (entry->GetId() != id))
{
if ((entry->Id() < objectId) &&
(entry->Id() > prevId)) prevId = entry->Id();
if ((entry->GetId() < objectId) &&
(entry->GetId() > prevId)) prevId = entry->GetId();
entry = entry->next;
}
if (entry) break;
}
} while (i != endex);
if (entry)
range_hi = entry->Id();
range_hi = entry->GetId();
else
range_hi = prevId;
range = range_hi - range_lo + 1;
@ -2583,7 +2582,7 @@ bool NormObjectTable::Remove(const NormObject* theObject)
range = 0;
}
count--;
size = size - theObject->Size();
size = size - theObject->GetSize();
return true;
}
else
@ -2633,15 +2632,15 @@ NormObject* NormObjectTable::Iterator::GetNextObject()
offset++;
NormObjectId id = (UINT16)index + offset;
NormObject* entry = table.table[i];
while ((NULL != entry) && (entry->Id() != id))
while ((NULL != entry) && (entry->GetId() != id))
{
if ((entry->Id() > index) && (entry->Id() < nextId))
nextId = entry->Id();
if ((entry->GetId() > index) && (entry->GetId() < nextId))
nextId = entry->GetId();
entry = table.Next(entry);
}
if (entry)
{
index = entry->Id();
index = entry->GetId();
return entry;
}
} while (i != endex);
@ -2692,15 +2691,15 @@ NormObject* NormObjectTable::Iterator::GetPrevObject()
offset--;
NormObjectId id = (UINT16)index + offset;
NormObject* entry = table.table[i];
while ((NULL != entry ) && (entry->Id() != id))
while ((NULL != entry ) && (entry->GetId() != id))
{
if ((entry->Id() > index) && (entry->Id() < nextId))
nextId = entry->Id();
if ((entry->GetId() > index) && (entry->GetId() < nextId))
nextId = entry->GetId();
entry = table.Next(entry);
}
if (entry)
{
index = entry->Id();
index = entry->GetId();
return entry;
}
} while (i != endex);

10
common/normObject.h
View File

@ -32,11 +32,11 @@ class NormObject
// Object information
NormObject::Type GetType() const {return type;}
const NormObjectId& Id() const {return id;}
const NormObjectSize& Size() const {return object_size;}
const NormObjectId& GetId() const {return transport_id;}
const NormObjectSize& GetSize() const {return object_size;}
bool HaveInfo() const {return (info_len > 0);}
const char* GetInfo() const {return info;}
UINT16 InfoLength() const {return info_len;}
UINT16 GetInfoLength() const {return info_len;}
bool IsStream() const {return (STREAM == type);}
NormNodeId LocalNodeId() const;
@ -127,7 +127,7 @@ class NormObject
NormSegmentId lastSegmentId,
UINT16 numErasures)
{
NormBlockId blockId = theBlock->Id();
NormBlockId blockId = theBlock->GetId();
bool result = theBlock->TxUpdate(firstSegmentId, lastSegmentId,
GetBlockSize(blockId), nparity,
numErasures);
@ -184,7 +184,7 @@ class NormObject
NormObject::Type type;
class NormSession* session;
class NormServerNode* server; // NULL value indicates local (tx) object
NormObjectId id;
NormObjectId transport_id;
NormObjectSize object_size;
UINT16 segment_size;

158
common/normSegment.cpp
View File

@ -104,28 +104,28 @@ NormBlock::~NormBlock()
Destroy();
}
bool NormBlock::Init(UINT16 blockSize)
bool NormBlock::Init(UINT16 totalSize)
{
if (segment_table) Destroy();
if (!(segment_table = new char*[blockSize]))
if (!(segment_table = new char*[totalSize]))
{
DMSG(0, "NormBlock::Init() segment_table allocation error: %s\n", strerror(errno));
return false;
}
memset(segment_table, 0, blockSize*sizeof(char*));
if (!pending_mask.Init(blockSize))
memset(segment_table, 0, totalSize*sizeof(char*));
if (!pending_mask.Init(totalSize))
{
DMSG(0, "NormBlock::Init() pending_mask allocation error: %s\n", strerror(errno));
Destroy();
return false;
}
if (!repair_mask.Init(blockSize))
if (!repair_mask.Init(totalSize))
{
DMSG(0, "NormBlock::Init() repair_mask allocation error: %s\n", strerror(errno));
Destroy();
return false;
}
size = blockSize;
size = totalSize;
erasure_count = 0;
parity_count = 0;
parity_offset = 0;;
@ -173,16 +173,16 @@ bool NormBlock::IsEmpty() const
// Used by client side to determine if NACK should be sent
// Note: This invalidates the block's "repair_mask" state
bool NormBlock::IsRepairPending(UINT16 ndata, UINT16 nparity)
bool NormBlock::IsRepairPending(UINT16 numData, UINT16 numParity)
{
// Clients ask for a block of parity to fulfill their
// repair needs (erasure_count), but if there isn't
// enough parity, they ask for some data segments, too
if (erasure_count > nparity)
if (erasure_count > numParity)
{
if (nparity)
if (numParity)
{
UINT16 i = nparity;
UINT16 i = numParity;
NormSegmentId nextId = 0;
GetFirstPending(nextId);
while (i--)
@ -194,44 +194,44 @@ bool NormBlock::IsRepairPending(UINT16 ndata, UINT16 nparity)
GetNextPending(nextId);
}
}
else if (size > ndata)
else if (size > numData)
{
repair_mask.SetBits(ndata, size-ndata);
repair_mask.SetBits(numData, size-numData);
}
}
else
{
repair_mask.SetBits(0, ndata);
repair_mask.SetBits(ndata+erasure_count, nparity-erasure_count);
repair_mask.SetBits(0, numData);
repair_mask.SetBits(numData+erasure_count, numParity-erasure_count);
}
repair_mask.XCopy(pending_mask);
return (repair_mask.IsSet());
} // end NormBlock::IsRepairPending()
// Called by server
bool NormBlock::TxReset(UINT16 ndata,
UINT16 nparity,
bool NormBlock::TxReset(UINT16 numData,
UINT16 numParity,
UINT16 autoParity,
UINT16 segmentSize)
{
bool increasedRepair = false;
repair_mask.SetBits(0, ndata+autoParity);
repair_mask.UnsetBits(ndata+autoParity, nparity-autoParity);
repair_mask.SetBits(0, numData+autoParity);
repair_mask.UnsetBits(numData+autoParity, numParity-autoParity);
repair_mask.Xor(pending_mask);
if (repair_mask.IsSet())
{
increasedRepair = true;
repair_mask.Clear();
pending_mask.SetBits(0, ndata+autoParity);
pending_mask.UnsetBits(ndata+autoParity, nparity-autoParity);
pending_mask.SetBits(0, numData+autoParity);
pending_mask.UnsetBits(numData+autoParity, numParity-autoParity);
parity_offset = autoParity; // reset parity since we're resending this one
parity_count = nparity; // no parity repair this repair cycle
parity_count = numParity; // no parity repair this repair cycle
SetFlag(IN_REPAIR);
if (!ParityReady(ndata)) // (TBD) only when incrementalParity == true
if (!ParityReady(numData)) // (TBD) only when incrementalParity == true
{
// Clear _any_ existing incremental parity state
char** ptr = segment_table+ndata;
while (nparity--)
char** ptr = segment_table+numData;
while (numParity--)
{
if (*ptr) memset(*ptr, 0, segmentSize);
ptr++;
@ -242,7 +242,7 @@ bool NormBlock::TxReset(UINT16 ndata,
return increasedRepair;
} // end NormBlock::TxReset()
bool NormBlock::ActivateRepairs(UINT16 nparity)
bool NormBlock::ActivateRepairs(UINT16 numParity)
{
if (repair_mask.IsSet())
{
@ -260,14 +260,14 @@ bool NormBlock::ActivateRepairs(UINT16 nparity)
// (we directly update the "pending_mask" for blocks/segments
// greater than our current transmit index)
bool NormBlock::TxUpdate(NormSegmentId nextId, NormSegmentId lastId,
UINT16 ndata, UINT16 nparity, UINT16 erasureCount)
UINT16 numData, UINT16 numParity, UINT16 erasureCount)
{
bool increasedRepair = false;
if (nextId < ndata)
if (nextId < numData)
{
// Explicit data repair request
parity_offset = parity_count = nparity;
parity_offset = parity_count = numParity;
while (nextId <= lastId)
{
if (!pending_mask.Test(nextId))
@ -281,13 +281,13 @@ bool NormBlock::TxUpdate(NormSegmentId nextId, NormSegmentId lastId,
else
{
// parity repair request
UINT16 parityAvailable = nparity - parity_offset;
UINT16 parityAvailable = numParity - parity_offset;
if (erasureCount <= parityAvailable)
{
// Use fresh parity for repair
if (erasureCount > parity_count)
{
pending_mask.SetBits(ndata+parity_offset+parity_count,
pending_mask.SetBits(numData+parity_offset+parity_count,
erasureCount - parity_count);
parity_count = erasureCount;
increasedRepair = true;
@ -299,7 +299,7 @@ bool NormBlock::TxUpdate(NormSegmentId nextId, NormSegmentId lastId,
if (parity_count < parityAvailable)
{
UINT16 count = parityAvailable - parity_count;
pending_mask.SetBits(ndata+parity_offset+parity_count, count);
pending_mask.SetBits(numData+parity_offset+parity_count, count);
parity_count = parityAvailable;
nextId += parityAvailable;
increasedRepair = true;
@ -320,15 +320,15 @@ bool NormBlock::TxUpdate(NormSegmentId nextId, NormSegmentId lastId,
} // end NormBlock::TxUpdate()
bool NormBlock::HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
UINT16 ndata, UINT16 nparity, UINT16 erasureCount)
UINT16 numData, UINT16 numParity, UINT16 erasureCount)
{
DMSG(6, "NormBlock::HandleSegmentRequest() blk>%lu seg>%hu:%hu erasures:%hu\n",
(UINT32)id, (UINT16)nextId, (UINT16)lastId, erasureCount);
bool increasedRepair = false;
if (nextId < ndata)
if (nextId < numData)
{
// Explicit data repair request
parity_count = parity_offset = nparity;
parity_count = parity_offset = numParity;
while (nextId <= lastId)
{
if (!repair_mask.Test(nextId))
@ -342,13 +342,13 @@ bool NormBlock::HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
else
{
// parity repair request
UINT16 parityAvailable = nparity - parity_offset;
UINT16 parityAvailable = numParity - parity_offset;
if (erasureCount <= parityAvailable)
{
// Use fresh parity for repair
if (erasureCount > parity_count)
{
repair_mask.SetBits(ndata+parity_offset+parity_count,
repair_mask.SetBits(numData+parity_offset+parity_count,
erasureCount - parity_count);
parity_count = erasureCount;
increasedRepair = true;
@ -360,7 +360,7 @@ bool NormBlock::HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
if (parity_count < parityAvailable)
{
UINT16 count = parityAvailable - parity_count;
repair_mask.SetBits(ndata+parity_offset+parity_count, count);
repair_mask.SetBits(numData+parity_offset+parity_count, count);
parity_count = parityAvailable;
nextId += parityAvailable;
increasedRepair = true;
@ -384,7 +384,7 @@ bool NormBlock::HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
NormObjectId objectId,
bool repairInfo,
UINT16 ndata,
UINT16 numData,
UINT16 segmentSize)
{
NormRepairRequest req;
@ -393,18 +393,18 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
NormSymbolId nextId = 0;
if (GetFirstRepair(nextId))
{
UINT16 blockSize = size;
UINT16 totalSize = size;
NormRepairRequest::Form prevForm = NormRepairRequest::INVALID;
UINT16 segmentCount = 0;
UINT16 firstId = 0;
while (nextId < blockSize)
while (nextId < totalSize)
{
UINT16 currentId = nextId;
if (!GetNextRepair(++nextId)) nextId = blockSize;
if (!GetNextRepair(++nextId)) nextId = totalSize;
if (!segmentCount) firstId = currentId;
segmentCount++;
// Check for break in consecutive series or end
if (((nextId - currentId) > 1) || (nextId >= blockSize))
if (((nextId - currentId) > 1) || (nextId >= totalSize))
{
NormRepairRequest::Form form;
switch (segmentCount)
@ -434,13 +434,13 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
ASSERT(0); // can't happen
break;
case NormRepairRequest::ITEMS:
req.AppendRepairItem(objectId, id, ndata, firstId);
req.AppendRepairItem(objectId, id, numData, firstId);
if (2 == segmentCount)
req.AppendRepairItem(objectId, id, ndata, currentId);
req.AppendRepairItem(objectId, id, numData, currentId);
break;
case NormRepairRequest::RANGES:
req.AppendRepairRange(objectId, id, ndata, firstId,
objectId, id, ndata, currentId);
req.AppendRepairRange(objectId, id, numData, firstId,
objectId, id, numData, currentId);
break;
case NormRepairRequest::ERASURES:
// erasure counts not used
@ -448,7 +448,7 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
}
segmentCount = 0;
}
} // end while (nextId < blockSize)
} // end while (nextId < totalSize)
if (NormRepairRequest::INVALID != prevForm)
cmd.PackRepairRequest(req); // (TBD) error check
}
@ -457,32 +457,32 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
// Called by client
bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
UINT16 ndata,
UINT16 nparity,
UINT16 numData,
UINT16 numParity,
NormObjectId objectId,
bool pendingInfo,
UINT16 segmentSize)
{
NormSegmentId nextId = 0;
NormSegmentId endId;
if (erasure_count > nparity)
if (erasure_count > numParity)
{
// Request explicit repair
GetFirstPending(nextId);
UINT16 i = nparity;
// Skip nparity missing data segments
UINT16 i = numParity;
// Skip numParity missing data segments
while (i--)
{
nextId++;
GetNextPending(nextId);
}
endId = ndata + nparity;
endId = numData + numParity;
}
else
{
nextId = ndata;
nextId = numData;
GetNextPending(nextId);
endId = ndata + erasure_count;
endId = numData + erasure_count;
}
NormRepairRequest req;
req.SetFlag(NormRepairRequest::SEGMENT);
@ -528,13 +528,13 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
ASSERT(0);
break;
case NormRepairRequest::ITEMS:
req.AppendRepairItem(objectId, id, ndata, firstId); // (TBD) error check
req.AppendRepairItem(objectId, id, numData, firstId); // (TBD) error check
if (2 == segmentCount)
req.AppendRepairItem(objectId, id, ndata, currentId); // (TBD) error check
req.AppendRepairItem(objectId, id, numData, currentId); // (TBD) error check
break;
case NormRepairRequest::RANGES:
req.AppendRepairRange(objectId, id, ndata, firstId, // (TBD) error check
objectId, id, ndata, currentId); // (TBD) error check
req.AppendRepairRange(objectId, id, numData, firstId, // (TBD) error check
objectId, id, numData, currentId); // (TBD) error check
break;
case NormRepairRequest::ERASURES:
// erasure counts not used
@ -627,7 +627,7 @@ NormBlockPool::~NormBlockPool()
Destroy();
}
bool NormBlockPool::Init(UINT32 numBlocks, UINT16 blockSize)
bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize)
{
if (head) Destroy();
for (UINT32 i = 0; i < numBlocks; i++)
@ -635,7 +635,7 @@ bool NormBlockPool::Init(UINT32 numBlocks, UINT16 blockSize)
NormBlock* b = new NormBlock();
if (b)
{
if (!b->Init(blockSize))
if (!b->Init(totalSize))
{
DMSG(0, "NormBlockPool::Init() block init error\n");
delete b;
@ -723,7 +723,7 @@ NormBlock* NormBlockBuffer::Find(const NormBlockId& blockId) const
if ((blockId < range_lo) || (blockId > range_hi))
return (NormBlock*)NULL;
NormBlock* theBlock = table[((UINT32)blockId) & hash_mask];
while (theBlock && (blockId != theBlock->Id()))
while (theBlock && (blockId != theBlock->GetId()))
theBlock = theBlock->next;
return theBlock;
}
@ -766,7 +766,7 @@ bool NormBlockBuffer::CanInsert(NormBlockId blockId) const
bool NormBlockBuffer::Insert(NormBlock* theBlock)
{
const NormBlockId& blockId = theBlock->Id();
const NormBlockId& blockId = theBlock->GetId();
if (!range)
{
range_lo = range_hi = blockId;
@ -789,7 +789,7 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock)
UINT32 index = ((UINT32)blockId) & hash_mask;
NormBlock* prev = NULL;
NormBlock* entry = table[index];
while (entry && (entry->Id() < blockId))
while (entry && (entry->GetId() < blockId))
{
prev = entry;
entry = entry->next;
@ -799,7 +799,7 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock)
else
table[index] = theBlock;
ASSERT((entry ? (blockId != entry->Id()) : true));
ASSERT((entry ? (blockId != entry->GetId()) : true));
theBlock->next = entry;
return true;
@ -810,12 +810,12 @@ bool NormBlockBuffer::Remove(const NormBlock* theBlock)
ASSERT(theBlock);
if (range)
{
const NormBlockId& blockId = theBlock->Id();
const NormBlockId& blockId = theBlock->GetId();
if ((blockId < range_lo) || (blockId > range_hi)) return false;
UINT32 index = ((UINT32)blockId) & hash_mask;
NormBlock* prev = NULL;
NormBlock* entry = table[index];
while (entry && (entry->Id() != blockId))
while (entry && (entry->GetId() != blockId))
{
prev = entry;
entry = entry->next;
@ -847,10 +847,10 @@ bool NormBlockBuffer::Remove(const NormBlock* theBlock)
if ((entry = table[i]))
{
NormBlockId id = (UINT32)index + offset;
while(entry && (entry->Id() != id))
while(entry && (entry->GetId() != id))
{
if ((entry->Id() > blockId) &&
(entry->Id() < nextId)) nextId = entry->Id();
if ((entry->GetId() > blockId) &&
(entry->GetId() < nextId)) nextId = entry->GetId();
entry = entry->next;
}
@ -858,7 +858,7 @@ bool NormBlockBuffer::Remove(const NormBlock* theBlock)
}
} while (i != endex);
if (entry)
range_lo = entry->Id();
range_lo = entry->GetId();
else
range_lo = nextId;
range = range_hi - range_lo + 1;
@ -884,17 +884,17 @@ bool NormBlockBuffer::Remove(const NormBlock* theBlock)
{
NormBlockId id = (UINT32)index - offset;
//printf("Looking for id:%lu at index:%lu\n", (UINT32)id, i);
while(entry && (entry->Id() != id))
while(entry && (entry->GetId() != id))
{
if ((entry->Id() < blockId) &&
(entry->Id() > prevId)) prevId = entry->Id();
if ((entry->GetId() < blockId) &&
(entry->GetId() > prevId)) prevId = entry->GetId();
entry = entry->next;
}
if (entry) break;
}
} while (i != endex);
if (entry)
range_hi = entry->Id();
range_hi = entry->GetId();
else
range_hi = prevId;
range = range_hi - range_lo + 1;
@ -954,15 +954,15 @@ NormBlock* NormBlockBuffer::Iterator::GetNextBlock()
NormBlockId id = (UINT32)index + offset;
ASSERT(i < 256);
NormBlock* entry = buffer.table[i];
while ((NULL != entry ) && (entry->Id() != id))
while ((NULL != entry ) && (entry->GetId() != id))
{
if ((entry->Id() > index) && (entry->Id() < nextId))
nextId = entry->Id();
if ((entry->GetId() > index) && (entry->GetId() < nextId))
nextId = entry->GetId();
entry = NormBlockBuffer::Next(entry);
}
if (entry)
{
index = entry->Id();
index = entry->GetId();
return entry;
}
} while (i != endex);

12
common/normSegment.h
View File

@ -56,9 +56,9 @@ class NormBlock
NormBlock();
~NormBlock();
const NormBlockId& Id() const {return id;}
const NormBlockId& GetId() const {return id;}
void SetId(NormBlockId& x) {id = x;}
bool Init(UINT16 blockSize);
bool Init(UINT16 totalSize);
void Destroy();
void SetFlag(NormBlock::Flag flag) {flags |= flag;}
@ -134,7 +134,7 @@ class NormBlock
bool AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
NormObjectId objectId,
bool repairInfo,
UINT16 ndata,
UINT16 numData,
UINT16 segmentSize);
// Client routines
@ -217,8 +217,8 @@ class NormBlock
{return (pending_mask.IsSet() || repair_mask.IsSet());}
bool AppendRepairRequest(NormNackMsg& nack,
UINT16 ndata,
UINT16 nparity,
UINT16 numData,
UINT16 numParity,
NormObjectId objectId,
bool pendingInfo,
UINT16 segmentSize);
@ -248,7 +248,7 @@ class NormBlockPool
public:
NormBlockPool();
~NormBlockPool();
bool Init(UINT32 numBlocks, UINT16 blockSize);
bool Init(UINT32 numBlocks, UINT16 totalSize);
void Destroy();
bool IsEmpty() const {return (NULL == head);}
NormBlock* Get()

162
common/normSession.cpp
View File

@ -21,7 +21,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
ttl(DEFAULT_TTL), tx_rate(DEFAULT_TRANSMIT_RATE/8.0),
backoff_factor(DEFAULT_BACKOFF_FACTOR), is_server(false), session_id(0),
ndata(DEFAULT_NDATA), nparity(DEFAULT_NPARITY), auto_parity(0), extra_parity(0),
next_tx_object_id(0), tx_cache_count_min(1), tx_cache_count_max(256),
next_tx_object_id(0), tx_cache_count_min(8), tx_cache_count_max(256),
tx_cache_size_max((UINT32)20*1024*1024),
flush_count(NORM_ROBUST_FACTOR+1),
posted_tx_queue_empty(false), advertise_repairs(false),
@ -72,7 +72,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
// (It may be used to trigger transmission of report messages
// in the future for debugging, etc
report_timer.SetListener(this, &NormSession::OnReportTimeout);
report_timer.SetInterval(10.0);
report_timer.SetInterval(60.0);
report_timer.SetRepeat(-1);
}
@ -233,6 +233,8 @@ bool NormSession::StartServer(unsigned long bufferSpace,
flush_count = NORM_ROBUST_FACTOR+1; // (TBD) parameterize robust_factor
//probe_timer.SetInterval(0.0);
probe_pending = probe_reset = false;
if (cc_enable) tx_rate = segmentSize;
OnProbeTimeout(probe_timer);
ActivateTimer(probe_timer);
return true;
@ -317,7 +319,7 @@ void NormSession::Serve()
if (obj->IsStream())
posted_tx_queue_empty = true; // repair-delayed stream advance
else
tx_pending_mask.Unset(obj->Id());
tx_pending_mask.Unset(obj->GetId());
}
}
else
@ -379,13 +381,13 @@ void NormSession::ServerQueueFlush()
if (obj->IsStream())
{
NormStreamObject* stream = (NormStreamObject*)obj;
objectId = stream->Id();
objectId = stream->GetId();
blockId = stream->FlushBlockId();
segmentId = stream->FlushSegmentId();
}
else
{
objectId = obj->Id();
objectId = obj->GetId();
blockId = obj->GetFinalBlockId();
segmentId = obj->GetBlockSize(blockId) - 1;
}
@ -579,7 +581,7 @@ bool NormSession::QueueTxObject(NormObject* obj, bool touchServer)
unsigned long count = tx_table.Count();
while ((count >= tx_cache_count_min) &&
((count >= tx_cache_count_max) ||
((tx_table.Size() + obj->Size()) > tx_cache_size_max)))
((tx_table.Size() + obj->GetSize()) > tx_cache_size_max)))
{
// Remove oldest non-pending
NormObject* oldest = tx_table.Find(tx_table.RangeLo());
@ -605,8 +607,8 @@ bool NormSession::QueueTxObject(NormObject* obj, bool touchServer)
ASSERT(0);
return false;
}
tx_pending_mask.Set(obj->Id());
ASSERT(tx_pending_mask.Test(obj->Id()));
tx_pending_mask.Set(obj->GetId());
ASSERT(tx_pending_mask.Test(obj->GetId()));
next_tx_object_id++;
TouchServer();
//posted_tx_queue_empty = false;
@ -616,7 +618,7 @@ bool NormSession::QueueTxObject(NormObject* obj, bool touchServer)
void NormSession::DeleteTxObject(NormObject* obj)
{
NormObjectId objectId = obj->Id();
NormObjectId objectId = obj->GetId();
ASSERT(obj == tx_table.Find(objectId));
tx_table.Remove(obj);
obj->Close();
@ -637,7 +639,7 @@ NormBlock* NormSession::ServerGetFreeBlock(NormObjectId objectId,
NormObject* obj;
while ((obj = iterator.GetNextObject()))
{
if (obj->Id() == objectId)
if (obj->GetId() == objectId)
b = obj->StealNonPendingBlock(true, blockId);
else
b = obj->StealNonPendingBlock(false);
@ -656,13 +658,13 @@ NormBlock* NormSession::ServerGetFreeBlock(NormObjectId objectId,
NormObject* obj;
while ((obj = iterator.GetPrevObject()))
{
if (obj->Id() < objectId)
if (obj->GetId() < objectId)
{
break;
}
else
{
if (obj->Id() > objectId)
if (obj->GetId() > objectId)
b = obj->StealNewestBlock(false);
else
b = obj->StealNewestBlock(true, blockId);
@ -888,34 +890,21 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
if (trace) NormTrace(currentTime, LocalNodeId(), msg, false);
// Do common updates for servers we already know.
NormNodeId sourceId = msg.GetSourceId();
NormServerNode* theServer = (NormServerNode*)server_tree.FindNodeById(sourceId);
if (theServer)
{
// (TBD) combine these as needed for efficiency
// (i.e. fewer function calls per message)
if (theServer->IsOpen()) theServer->Activate();
theServer->UpdateRecvRate(currentTime, msg.GetLength());
theServer->UpdateLossEstimate(currentTime, msg.GetSequence());
theServer->SetAddress(msg.GetSource());
// for statistics only (TBD) #ifdef NORM_DEBUG
theServer->IncrementRecvTotal(msg.GetLength());
}
switch (msg.GetType())
{
case NormMsg::INFO:
//DMSG(0, "NormSession::HandleReceiveMessage(NormMsg::INFO)\n");
if (IsClient()) ClientHandleObjectMessage(currentTime, (NormObjectMsg&)msg, theServer);
if (IsClient()) ClientHandleObjectMessage(currentTime, (NormObjectMsg&)msg);
break;
case NormMsg::DATA:
//DMSG(0, "NormSession::HandleReceiveMessage(NormMsg::DATA) ...\n");
if (IsClient()) ClientHandleObjectMessage(currentTime, (NormObjectMsg&)msg, theServer);
if (IsClient()) ClientHandleObjectMessage(currentTime, (NormObjectMsg&)msg);
break;
case NormMsg::CMD:
//DMSG(0, "NormSession::HandleReceiveMessage(NormMsg::CMD) ...\n");
if (IsClient()) ClientHandleCommand(currentTime, (NormCmdMsg&)msg, theServer);
if (IsClient()) ClientHandleCommand(currentTime, (NormCmdMsg&)msg);
break;
case NormMsg::NACK:
DMSG(4, "NormSession::HandleReceiveMessage(NormMsg::NACK) node>%lu ...\n",
@ -923,7 +912,7 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
if (IsServer() && (((NormNackMsg&)msg).GetServerId() == LocalNodeId()))
{
ServerHandleNackMessage(currentTime, (NormNackMsg&)msg);
if (wasUnicast && (backoff_factor > 0.5))
if (wasUnicast && (backoff_factor > 0.5) && Address().IsMulticast())
{
// for suppression of unicast nack feedback
advertise_repairs = true;
@ -952,9 +941,11 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
void NormSession::ClientHandleObjectMessage(const struct timeval& currentTime,
const NormObjectMsg& msg,
NormServerNode* theServer)
const NormObjectMsg& msg)
{
// Do common updates for servers we already know.
NormNodeId sourceId = msg.GetSourceId();
NormServerNode* theServer = (NormServerNode*)server_tree.FindNodeById(sourceId);
if (!theServer)
{
if ((theServer = new NormServerNode(this, msg.GetSourceId())))
@ -962,11 +953,6 @@ void NormSession::ClientHandleObjectMessage(const struct timeval& currentTime,
server_tree.AttachNode(theServer);
DMSG(4, "NormSession::ClientHandleObjectMessage() node>%lu new remote server:%lu ...\n",
LocalNodeId(), msg.GetSourceId());
theServer->UpdateRecvRate(currentTime, msg.GetLength());
theServer->UpdateLossEstimate(currentTime, msg.GetSequence());
theServer->SetAddress(msg.GetSource());
// for statistics only (TBD) #ifdef NORM_DEBUG
theServer->IncrementRecvTotal(msg.GetLength());
}
else
{
@ -976,13 +962,20 @@ void NormSession::ClientHandleObjectMessage(const struct timeval& currentTime,
return;
}
}
if (theServer->IsOpen()) theServer->Activate();
theServer->UpdateLossEstimate(currentTime, msg.GetSequence());
theServer->SetAddress(msg.GetSource());
theServer->IncrementRecvTotal(msg.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG
theServer->HandleObjectMessage(msg);
theServer->UpdateRecvRate(currentTime, msg.GetLength());
} // end NormSession::ClientHandleObjectMessage()
void NormSession::ClientHandleCommand(const struct timeval& currentTime,
const NormCmdMsg& cmd,
NormServerNode* theServer)
const NormCmdMsg& cmd)
{
// Do common updates for servers we already know.
NormNodeId sourceId = cmd.GetSourceId();
NormServerNode* theServer = (NormServerNode*)server_tree.FindNodeById(sourceId);
if (!theServer)
{
//DMSG(0, "NormSession::ClientHandleCommand() node>%lu recvd command from unknown server ...\n",
@ -993,11 +986,6 @@ void NormSession::ClientHandleCommand(const struct timeval& currentTime,
server_tree.AttachNode(theServer);
DMSG(4, "NormSession::ClientHandleCommand() node>%lu new remote server:%lu ...\n",
LocalNodeId(), cmd.GetSourceId());
theServer->UpdateRecvRate(currentTime, cmd.GetLength());
theServer->UpdateLossEstimate(currentTime, cmd.GetSequence());
theServer->SetAddress(cmd.GetSource());
// for statistics only (TBD) #ifdef NORM_DEBUG
theServer->IncrementRecvTotal(cmd.GetLength());
}
else
{
@ -1007,7 +995,12 @@ void NormSession::ClientHandleCommand(const struct timeval& currentTime,
return;
}
}
if (theServer->IsOpen()) theServer->Activate();
theServer->UpdateLossEstimate(currentTime, cmd.GetSequence());
theServer->SetAddress(cmd.GetSource());
theServer->IncrementRecvTotal(cmd.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG
theServer->HandleCommand(currentTime, cmd);
theServer->UpdateRecvRate(currentTime, cmd.GetLength());
} // end NormSession::ClientHandleCommand()
@ -1046,16 +1039,16 @@ double NormSession::CalculateRtt(const struct timeval& currentTime,
void NormSession::ServerUpdateGrttEstimate(double clientRtt)
{
grtt_response = true;
if (clientRtt > grtt_current_peak)
if ((clientRtt > grtt_current_peak) || !address.IsMulticast())
{
// Immediately incorporate bigger RTT's
grtt_current_peak = clientRtt;
if (clientRtt > grtt_measured)
if ((clientRtt > grtt_measured) || !address.IsMulticast())
{
grtt_decrease_delay_count = DEFAULT_GRTT_DECREASE_DELAY;
grtt_measured = 0.25 * grtt_measured + 0.75 * clientRtt;
if (grtt_measured > grtt_max) grtt_measured = grtt_max;
double pktInterval = 2.0 * ((double)(44+segment_size))/tx_rate;
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
@ -1104,6 +1097,7 @@ void NormSession::ServerHandleCCFeedback(NormNodeId nodeId,
// Adjust ccRtt if we have state on this nodeId
NormCCNode* node = (NormCCNode*)cc_node_list.FindNodeById(nodeId);
if (node) ccRtt = node->UpdateRtt(ccRtt);
if (0 == (ccFlags & NormCC::START))
{
// slow start has ended
@ -1287,7 +1281,7 @@ void NormSession::ServerHandleAckMessage(const struct timeval& currentTime, cons
break;
}
if (wasUnicast && probe_proactive)
if (wasUnicast && probe_proactive && Address().IsMulticast())
{
// for suppression of unicast feedback
advertise_repairs = true;
@ -1711,7 +1705,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
else
{
block->HandleSegmentRequest(nextSegmentId, lastSegmentId,
object->GetBlockSize(block->Id()),
object->GetBlockSize(block->GetId()),
nparity, numErasures);
startTimer = true;
} // end if/else (holdoff)
@ -1798,7 +1792,7 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId)
squelch->SetFecSymbolId(0);
squelch->ResetInvalidObjectList();
while ((obj = iterator.GetNextObject()))
if (objectId == obj->Id()) break;
if (objectId == obj->GetId()) break;
nextId = objectId + 1;
}
else
@ -1806,13 +1800,13 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId)
obj = iterator.GetNextObject();
if (obj)
{
squelch->SetObjectId(obj->Id());
squelch->SetObjectId(obj->GetId());
if (obj->IsStream())
squelch->SetFecBlockId(((NormStreamObject*)obj)->StreamBufferLo());
else
squelch->SetFecBlockId(0);
squelch->SetFecSymbolId(0);
nextId = obj->Id() + 1;
nextId = obj->GetId() + 1;
}
else
{
@ -1826,7 +1820,7 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId)
bool buildingList = true;
while (buildingList && (obj = iterator.GetNextObject()))
{
while (nextId != obj->Id())
while (nextId != obj->GetId())
{
if (!squelch->AppendInvalidObject(nextId, segment_size))
{
@ -1864,7 +1858,7 @@ bool NormSession::ServerBuildRepairAdv(NormCmdRepairAdvMsg& cmd)
{
NormObject* currentObject = nextObject;
nextObject = iterator.GetNextObject();
NormObjectId currentId = currentObject->Id();
NormObjectId currentId = currentObject->GetId();
bool repairEntireObject = tx_repair_mask.Test(currentId);
if (repairEntireObject)
{
@ -1953,7 +1947,7 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/)
NormObject* obj;
while ((obj = iterator.GetNextObject()))
{
NormObjectId objectId = obj->Id();
NormObjectId objectId = obj->GetId();
if (tx_repair_mask.Test(objectId))
{
DMSG(6, "NormSession::OnRepairTimeout() node>%lu tx reset obj>%hu ...\n",
@ -2167,23 +2161,17 @@ bool NormSession::SendMessage(NormMsg& msg)
// Separate send/recv tracing
if (trace) NormTrace(currentTime, LocalNodeId(), msg, true);
// Keep track of _actual_ sent rate
double interval;
// Keep track of _actual_ sent rate (TBD) we can do away with this now
if (prev_update_time.tv_sec || prev_update_time.tv_usec)
{
interval = (double)(currentTime.tv_sec - prev_update_time.tv_sec);
double interval = (double)(currentTime.tv_sec - 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);
else
interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec);
}
else
{
sent_rate = ((double)msgSize) / grtt_advertised;
interval = -1.0;
sent_rate += 0.125 * (((double)msgSize)/interval - sent_rate);
prev_update_time = currentTime;
}
if (interval < grtt_advertised)
/*if (interval < grtt_advertised)
{
sent_accumulator += msgSize;
}
@ -2192,6 +2180,13 @@ bool NormSession::SendMessage(NormMsg& msg)
sent_rate = ((double)(sent_accumulator+msgSize)) / interval;
prev_update_time = currentTime;
sent_accumulator = 0;
}*/
}
else
{
sent_rate = ((double)msgSize) / grtt_advertised;
prev_update_time = currentTime;
sent_accumulator = msgSize;
}
// Update nominal packet size
nominal_packet_size += 0.05 * (((double)msgSize) - nominal_packet_size);
@ -2289,6 +2284,7 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/)
cmd->Init();
cmd->SetDestination(address);
cmd->SetGrtt(grtt_quantized);
cmd->SetBackoffFactor((unsigned char)backoff_factor);
cmd->SetGroupSize(gsize_quantized);
// SetSendTime() when message is being sent (in OnTxTimeout())
@ -2332,8 +2328,8 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/)
// Determine next probe_interval
const NormCCNode* clr = (const NormCCNode*)cc_node_list.Head();
probeInterval = clr ? MIN(grtt_advertised, clr->GetRtt()) : grtt_advertised;
double nominalRate = ((double)segment_size)/((double)tx_rate);
probeInterval = MAX(probeInterval, nominalRate);
//double nominalRate = ((double)segment_size)/((double)tx_rate);
//probeInterval = MAX(probeInterval, nominalRate);
}
else
{
@ -2356,6 +2352,7 @@ void NormSession::AdjustRate(bool onResponse)
const NormCCNode* clr = (const NormCCNode*)cc_node_list.Head();
double ccRtt = clr ? clr->GetRtt() : grtt_measured;
double ccLoss = clr ? clr->GetLoss() : 0.0;
double oldRate = tx_rate;
if (onResponse)
{
// Adjust rate based on CLR feedback and
@ -2364,19 +2361,16 @@ void NormSession::AdjustRate(bool onResponse)
// (TBD) check feedback age
if (cc_slow_start)
{
double scale = tx_rate / sent_rate;
scale = MAX(1.0, scale);
scale = 1.0;
//DMSG(0, "NormSession::AdjustRate() slow start clr>%lu rate>%lf tx_rate>%lf sent_rate>%lf\n",
// clr->Id(), clr->GetRate() * 8.0/1000.0, tx_rate * 8.0/1000.0, sent_rate * 8.0/1000.0);
tx_rate = clr->GetRate() * scale;
tx_rate = clr->GetRate();
DMSG(6, "NormSession::AdjustRate(slow start) clr>%lu newRate>%lf (oldRate>%lf sentRate>%lf clrRate>%lf\n",
clr->GetId(), tx_rate*8.0/1000.0, oldRate*8.0/1000.0, sent_rate*8.0/1000.0,
clr->GetRate()*8.0/1000.0);
}
else
{
double clrRate = clr->GetRate();
if (clrRate > tx_rate)
{
double linRate = tx_rate + segment_size;
tx_rate = MIN(clrRate, linRate);
}
@ -2384,9 +2378,9 @@ void NormSession::AdjustRate(bool onResponse)
{
tx_rate = clrRate;
}
DMSG(6, "NormSession::AdjustRate(stdy state) clr>%lu newRate>%lf (rtt>%lf loss>%lf)\n",
clr->GetId(), tx_rate*8.0/1000.0, clr->GetRtt(), clr->GetLoss());
}
//DMSG(0, "NormSession::AdjustRate() regular rate adjust clr>%lu rate>%lf (rtt>%lf loss>%lf slow_start>%d)\n",
// clr->Id(), tx_rate*8.0/1000.0, clr->GetRtt(), clr->GetLoss(), cc_slow_start);
}
else if (clr)
{
@ -2414,7 +2408,21 @@ void NormSession::AdjustRate(bool onResponse)
struct timeval currentTime;
::ProtoSystemTime(currentTime);
double theTime = (double)currentTime.tv_sec + 1.0e-06 * ((double)currentTime.tv_usec);
DMSG(6, "ServerRateTracking time>%lf rate>%lf rtt>%lf loss>%lf\n", theTime, tx_rate * (8.0/1000.0), ccRtt, ccLoss);
if (tx_rate != oldRate)
{
if (tx_timer.IsActive())
{
//double ratio = tx_rate / oldRate;
double txInterval = tx_timer.GetInterval() * oldRate / tx_rate;
double timeElapsed = tx_timer.GetInterval() - tx_timer.GetTimeRemaining();
txInterval = timeElapsed < txInterval ? (txInterval - timeElapsed) : 0.0;
tx_timer.SetInterval(txInterval);
tx_timer.Reschedule();
}
}
DMSG(6, "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*/)

14
common/normSession.h
View File

@ -162,6 +162,14 @@ class NormSession
return result;
}
double ServerGrtt() {return grtt_advertised;}
void ServerSetGrtt(double grttValue)
{
double grttMin = 2.0 * ((double)(44+segment_size))/tx_rate;
grttValue = (grttValue < grttMin) ? grttMin : grttValue;
grtt_quantized = NormQuantizeRtt(grttValue);
grtt_measured = grtt_advertised = NormUnquantizeRtt(grtt_quantized);
}
double ServerGroupSize() {return gsize_measured;}
void ServerSetGroupSize(double gsize)
{
@ -267,11 +275,9 @@ class NormSession
// Client message handling routines
void ClientHandleObjectMessage(const struct timeval& currentTime,
const NormObjectMsg& msg,
NormServerNode* theServer);
const NormObjectMsg& msg);
void ClientHandleCommand(const struct timeval& currentTime,
const NormCmdMsg& msg,
NormServerNode* theServer);
const NormCmdMsg& msg);
void ClientHandleNackMessage(const NormNackMsg& nack);
void ClientHandleAckMessage(const NormAckMsg& ack);

62
common/normSimAgent.cpp
View File

@ -15,8 +15,10 @@ NormSimAgent::NormSimAgent(ProtoTimerMgr& timerMgr,
backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR),
segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0),
group_size(NormSession::DEFAULT_GSIZE_ESTIMATE),
grtt_estimate(NormSession::DEFAULT_GRTT_ESTIMATE),
tx_buffer_size(1024*1024), rx_buffer_size(1024*1024),
tx_object_size(0), tx_object_interval(0.0),
tx_object_size_min(0), tx_object_size_max(0),
tx_repeat_count(0), tx_repeat_interval(0.0),
stream(NULL), auto_stream(false), push_stream(false),
flush_mode(NormStreamObject::FLUSH_PASSIVE),
@ -62,11 +64,13 @@ const char* const NormSimAgent::cmd_list[] =
"+auto", // server auto parity count
"+extra", // server extra parity count
"+gsize", // group size estimate
"+grtt", // grtt estimate
"+txbuffer", // tx buffer size (bytes)
"+rxbuffer", // rx buffer size (bytes)
"+start", // open session and beging activity
"-stop", // cease activity and close session
"+sendFile", // queue a "sim" file for transmission
"+sendFile", // queue a "sim" file of <size> bytes for transmission
"+sendRandomFile", // queue random-size file size range <sizeMin>:<sizeMax>
"+sendStream", // send a simulated NORM stream
"+openStream", // open a stream object for messaging
"+push", // "on" means real-time push stream advancement (non-blocking)
@ -285,6 +289,15 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
}
if (session) session->ServerSetGroupSize(group_size);
}
else if (!strncmp("grtt", cmd, len))
{
if (1 != sscanf(val, "%lf", &grtt_estimate))
{
DMSG(0, "NormSimAgent::ProcessCommand(gize) invalid value!\n");
return false;
}
if (session) session->ServerSetGroupSize(grtt_estimate);
}
else if (!strncmp("txbuffer", cmd, len))
{
if (1 != sscanf(val, "%lu", &tx_buffer_size))
@ -338,6 +351,33 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
return false;
}
}
else if (!strncmp("sendRandomFile", cmd, len))
{
if (2 != sscanf(val, "%lu:%lu", &tx_object_size_min, &tx_object_size_max))
{
DMSG(0, "NormSimAgent::ProcessCommand(sendRandomFile) invalid size!\n");
return false;
}
if (tx_object_size_min > tx_object_size_max)
{
unsigned int tempSize = tx_object_size_min;
tx_object_size_min = tx_object_size_max;
tx_object_size_max = tempSize;
}
tx_object_size = tx_object_size_max - tx_object_size_min;
tx_object_size = (unsigned long)(tx_object_size * UniformRand(1.0));
tx_object_size += tx_object_size_min;
if (session)
{
TRACE("Queued file size: %lu bytes\n", tx_object_size);
return (NULL != session->QueueTxSim(tx_object_size));
}
else
{
DMSG(0, "NormSimAgent::ProcessCommand(sendRandomFile) no session started!\n");
return false;
}
}
else if (!strncmp("sendStream", cmd, len))
{
if (session)
@ -422,7 +462,7 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
}
else
{
DMSG(0, "NormSimAgent::ProcessCommand(flushStream) session not started!\n");
DMSG(0, "NormSimAgent::ProcessCommand(doFlush) session not started!\n");
return false;
}
}
@ -532,8 +572,8 @@ bool NormSimAgent::SendMessage(unsigned int len, const char* txBuffer)
else
{
// Message still pending, can't send yet
DMSG(0, "NormSimAgent::SendMessage() input overflow!\n");
ASSERT(0);
DMSG(0, "NormSimAgent::SendMessage() warning: input overflow!\n");
//ASSERT(0);
return false;
}
}
@ -564,11 +604,12 @@ void NormSimAgent::OnInputReady()
}
} // end NormSimAgent::InputReady()
// Do an _active_ flush of the stream (for use at end-of-transmission)
bool NormSimAgent::FlushStream()
{
if (stream && session && session->IsServer())
{
stream->Write(NULL, 0, flush_mode, false, false);
stream->Write(NULL, 0, NormStreamObject::FLUSH_ACTIVE, false, false);
return true;
}
else
@ -631,7 +672,7 @@ void NormSimAgent::Notify(NormController::Event event,
if (silent_client)
size = NormObjectSize(rx_buffer_size);
else
size = object->Size();
size = object->GetSize();
if (!((NormStreamObject*)object)->Accept(size.LSB()))
{
DMSG(0, "NormSimAgent::Notify(RX_OBJECT_NEW) stream object accept error!\n");
@ -823,10 +864,17 @@ bool NormSimAgent::OnIntervalTimeout(ProtoTimer& theTimer)
{
// Queue a NORM_OBJECT_SIM as long as there are repeats
if (tx_repeat_count > 0) tx_repeat_count--;
if (tx_object_size_min > 0)
{
tx_object_size = tx_object_size_max - tx_object_size_min;
tx_object_size = (unsigned long)(tx_object_size * UniformRand(1.0));
tx_object_size += tx_object_size_min;
}
if (!session->QueueTxSim(tx_object_size))
{
DMSG(0, "NormSimAgent::OnIntervalTimeout() Error queueing tx object.\n");
}
TRACE("Queued file size: %lu bytes\n", tx_object_size);
}
interval_timer.SetInterval(tx_object_interval);
return true;
@ -837,7 +885,6 @@ bool NormSimAgent::OnIntervalTimeout(ProtoTimer& theTimer)
if (interval_timer.IsActive()) interval_timer.Deactivate();
return false;
}
} // end NormSimAgent::OnIntervalTimeout()
@ -867,6 +914,7 @@ bool NormSimAgent::StartServer()
session->SetTxLoss(tx_loss);
session->SetRxLoss(rx_loss);
session->ServerSetGroupSize(group_size);
session->ServerSetGrtt(grtt_estimate);
// StartServer(bufferMax, segmentSize, fec_ndata, fec_nparity)
if (!session->StartServer(tx_buffer_size, segment_size, ndata, nparity))
{

3
common/normSimAgent.h
View File

@ -64,12 +64,15 @@ class NormSimAgent : public NormController
UINT8 auto_parity;
UINT8 extra_parity;
double group_size;
double grtt_estimate;
unsigned long tx_buffer_size; // bytes
unsigned long rx_buffer_size; // bytes
// for simulated transmission (streams or files)
unsigned long tx_object_size;
double tx_object_interval;
unsigned long tx_object_size_min;
unsigned long tx_object_size_max;
int tx_repeat_count;
double tx_repeat_interval;

2
common/normVersion.h
View File

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