diff --git a/common/normApp.cpp b/common/normApp.cpp index 84fb340..eb7e132 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -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 diff --git a/common/normNode.cpp b/common/normNode.cpp index 8b3410c..8dc4a3f 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -262,49 +262,51 @@ 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 - double backoffTime = (maxBackoff > 0.0) ? - ExponentialRand(maxBackoff, gsize_estimate) : 0.0; - // Bias backoff timeout based on our rate - double r; - if (slow_start) - { - r = recv_rate / send_rate; - cc_rate = 2.0 * recv_rate; - } - else - { - cc_rate = NormSession::CalculateRate(nominal_packet_size, - rtt_estimate, - LossEstimate()); - r = cc_rate / send_rate; - r = MIN(r, 0.9); - 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 " - // "grtt:%lf gsize:%lf\n", - // LocalNodeId(), r, recv_rate*(8.0/1000.0), send_rate*(8.0/1000.0), - // + maxBackoff = grtt_estimate*backoffFactor; - 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", - LocalNodeId(), backoffTime); - session->ActivateTimer(cc_timer); } + double backoffTime = (maxBackoff > 0.0) ? + ExponentialRand(maxBackoff, gsize_estimate) : + 0.0; + // Bias backoff timeout based on our rate + double r; + if (slow_start) + { + r = recv_rate / send_rate; + cc_rate = 2.0 * recv_rate; + } + else + { + cc_rate = NormSession::CalculateRate(nominal_packet_size, + rtt_estimate, + LossEstimate()); + r = cc_rate / send_rate; + r = MIN(r, 0.9); + r = MAX(r, 0.5); + r = (r - 0.5) / 0.4; + } + //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(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) - { + interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec); 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); diff --git a/common/normObject.cpp b/common/normObject.cpp index 615b414..955751f 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -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); diff --git a/common/normObject.h b/common/normObject.h index 1579e0b..1b16d20 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -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; diff --git a/common/normSegment.cpp b/common/normSegment.cpp index 5990a07..dbc1af3 100644 --- a/common/normSegment.cpp +++ b/common/normSegment.cpp @@ -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); diff --git a/common/normSegment.h b/common/normSegment.h index c824458..dbb1268 100644 --- a/common/normSegment.h +++ b/common/normSegment.h @@ -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() diff --git a/common/normSession.cpp b/common/normSession.cpp index 4de54cf..d1fd6ee 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -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,31 +2161,32 @@ 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); + interval -= 1.0e-06*(double)(prev_update_time.tv_usec - currentTime.tv_usec); + sent_rate += 0.125 * (((double)msgSize)/interval - sent_rate); + prev_update_time = currentTime; + /*if (interval < grtt_advertised) + { + sent_accumulator += msgSize; + } + else + { + sent_rate = ((double)(sent_accumulator+msgSize)) / interval; + prev_update_time = currentTime; + sent_accumulator = 0; + }*/ } else { sent_rate = ((double)msgSize) / grtt_advertised; - interval = -1.0; - prev_update_time = currentTime; - } - if (interval < grtt_advertised) - { - sent_accumulator += msgSize; - } - else - { - sent_rate = ((double)(sent_accumulator+msgSize)) / interval; - prev_update_time = currentTime; - sent_accumulator = 0; + prev_update_time = currentTime; + sent_accumulator = msgSize; } // Update nominal packet size nominal_packet_size += 0.05 * (((double)msgSize) - nominal_packet_size); @@ -2224,7 +2219,7 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) probe_reset = true; probe_timer.Deactivate(); return false; - } + } // 2) Update grtt_estimate _if_ sufficient time elapsed. // This new code allows more liberal downward adjustment of @@ -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*/) diff --git a/common/normSession.h b/common/normSession.h index 89074a3..0fc9726 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -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); diff --git a/common/normSimAgent.cpp b/common/normSimAgent.cpp index cb5ccbf..7e81900 100644 --- a/common/normSimAgent.cpp +++ b/common/normSimAgent.cpp @@ -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 bytes for transmission + "+sendRandomFile", // queue random-size file size range : "+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"); @@ -822,11 +863,18 @@ bool NormSimAgent::OnIntervalTimeout(ProtoTimer& theTimer) else { // Queue a NORM_OBJECT_SIM as long as there are repeats - if (tx_repeat_count > 0) tx_repeat_count--; + 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() @@ -866,7 +913,8 @@ bool NormSimAgent::StartServer() session->SetTrace(tracing); session->SetTxLoss(tx_loss); session->SetRxLoss(rx_loss); - session->ServerSetGroupSize(group_size); + 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)) { diff --git a/common/normSimAgent.h b/common/normSimAgent.h index fbfb2d6..319938d 100644 --- a/common/normSimAgent.h +++ b/common/normSimAgent.h @@ -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; diff --git a/common/normVersion.h b/common/normVersion.h index 1bc9b5b..f4ab427 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.1b5" +#define VERSION "1.1b6" #endif // _NORM_VERSION