diff --git a/CMakeLists.txt b/CMakeLists.txt index c0126ad..f7ef367 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -12,6 +12,17 @@ set(COMMON src/common) option(NORM_BUILD_EXAMPLES "Enables building of the examples in /examples." OFF) set(NORM_CUSTOM_PROTOLIB_VERSION OFF CACHE STRING "Set a custom protolib version to use, ./protolib to use the local version") +# Build self-checking tests by default only when NORM is the top-level project. +if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR) + set(NORM_BUILD_TESTING_DEFAULT ON) +else() + set(NORM_BUILD_TESTING_DEFAULT OFF) +endif() +option(NORM_BUILD_TESTING "Enables building NORM self-checking tests." ${NORM_BUILD_TESTING_DEFAULT}) +if(NORM_BUILD_TESTING) + enable_testing() +endif() + include(CheckCXXSymbolExists) check_cxx_symbol_exists(dirfd "dirent.h" HAVE_DIRFD) if(HAVE_DIRFD) @@ -224,3 +235,20 @@ if(NORM_BUILD_EXAMPLES) endforeach() endif() +if(NORM_BUILD_TESTING) + # This test exercises internal sender repair state and message generation, + # so compile it with the implementation sources instead of depending on + # symbols hidden by shared-library builds. + add_executable(normRepairFlagTest + tests/normRepairFlagTest.cpp + ${PLATFORM_SOURCE_FILES} + ${COMMON_SOURCE_FILES}) + target_link_libraries(normRepairFlagTest PRIVATE protokit::protokit ${PLATFORM_LIBS}) + target_compile_definitions(normRepairFlagTest PRIVATE ${PLATFORM_DEFINITIONS}) + target_compile_options(normRepairFlagTest PRIVATE ${PLATFORM_FLAGS}) + target_include_directories(normRepairFlagTest PRIVATE + ${CMAKE_CURRENT_BINARY_DIR} + ${CMAKE_CURRENT_SOURCE_DIR}/include) + add_test(NAME normRepairFlagTest COMMAND normRepairFlagTest) + set_tests_properties(normRepairFlagTest PROPERTIES TIMEOUT 15) +endif() diff --git a/TODO.TXT b/TODO.TXT index 5eb5870..cd754c8 100755 --- a/TODO.TXT +++ b/TODO.TXT @@ -1,7 +1,5 @@ TODO: -0) Fix implementation of NORM_DATA_FLAG_REPAIR. - 2) Add mechanism to allow window-based control of NORM server transmission rate. diff --git a/VERSION.TXT b/VERSION.TXT index 8bba414..e0f4bfb 100755 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -4,6 +4,9 @@ Version 1.5.10 (in progress) ============= - NormSocket API improvements + - Corrected repair flagging for reactive NORM_DATA and NORM_INFO + transmissions so receiver synchronization policy can distinguish + repair-only traffic from first-pass content. Version 1.5.9 ============= diff --git a/include/normObject.h b/include/normObject.h index 5b22945..6408961 100755 --- a/include/normObject.h +++ b/include/normObject.h @@ -89,12 +89,14 @@ class NormObject info_ptr = NULL; info_len = 0; pending_info = false; + pending_info_repair = false; } // This is used to bootstrap reception sender // is using FTI_INFO FtiMode void SetPendingInfo(bool state, UINT8 fecId) { pending_info = true; + pending_info_repair = false; fec_id = fecId; } @@ -318,6 +320,7 @@ class NormObject UINT16 nparity; NormBlockBuffer block_buffer; bool pending_info; // set when we need to send or recv info + bool pending_info_repair; // sender: pending INFO is reactive repair ProtoSlidingMask pending_mask; bool repair_info; // receiver: set when ProtoSlidingMask repair_mask; diff --git a/include/normSegment.h b/include/normSegment.h index cf88173..7b7a423 100755 --- a/include/normSegment.h +++ b/include/normSegment.h @@ -114,6 +114,7 @@ class NormBlock parity_offset = autoParity; flags = 0; seg_size_max = 0; + TxResetOriginalPass(ndata, autoParity); last_nack_time.GetCurrentTime(); } void TxRecover(NormBlockId& blockId, UINT16 ndata, UINT16 nparity) @@ -126,6 +127,32 @@ class NormBlock parity_offset = nparity; // explicit repair mode ??? flags = IN_REPAIR; seg_size_max = 0; + original_tx_next = original_tx_end = ndata; + } + void TxResetOriginalPass(UINT16 ndata, UINT16 autoParity) + { + ASSERT((ndata + autoParity) <= size); + original_tx_next = 0; + original_tx_end = ndata + autoParity; + } + void TxCompleteOriginalPass() + { + original_tx_next = original_tx_end; + } + bool TxIsOriginalPending(NormSegmentId segmentId) const + { + return ((segmentId >= original_tx_next) && + (segmentId < original_tx_end)); + } + void TxAdvanceOriginal(NormSegmentId segmentId) + { + if (TxIsOriginalPending(segmentId)) + { + // Sender selection is ordered, so first-pass symbols cannot + // skip ahead of this frontier. + ASSERT(segmentId == original_tx_next); + original_tx_next = segmentId + 1; + } } bool TxReset(UINT16 ndata, UINT16 nparity, UINT16 autoParity, UINT16 segmentSize); @@ -164,6 +191,7 @@ class NormBlock parity_count = 0; parity_offset = 0; flags = 0; + original_tx_next = original_tx_end = 0; } // Note: This invalidates the repair_mask state. bool IsRepairPending(UINT16 ndata, UINT16 nparity); @@ -279,6 +307,8 @@ class NormBlock UINT16 parity_count; // how many fresh parity we are currently planning to send UINT16 parity_offset; // offset from where our fresh parity will be sent UINT16 seg_size_max; + NormSegmentId original_tx_next; // first unsent symbol in the original pass + NormSegmentId original_tx_end; // source data plus proactive parity ProtoBitmask pending_mask; ProtoBitmask repair_mask; diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index fdda3da..2c90ea6 100755 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -12,7 +12,7 @@ NormObject::NormObject(NormObject::Type theType, class NormSenderNode* theSender, const NormObjectId& transportId) : type(theType), session(theSession), sender(theSender), reference_count(1), - transport_id(transportId), segment_size(0), pending_info(false), repair_info(false), + transport_id(transportId), segment_size(0), pending_info(false), pending_info_repair(false), repair_info(false), current_block_id(0), next_segment_id(0), max_pending_block(0), max_pending_segment(0), info_ptr(NULL), info_len(0), first_pass(true), accepted(false), notify_on_update(true), @@ -82,6 +82,7 @@ bool NormObject::Open(const NormObjectSize& objectSize, UINT16 numData, UINT16 numParity) { + pending_info_repair = false; // Note "objectSize" represents actual total object size for // DATA or FILE objects, buffer size for STREAM objects // In either case, we need our sliding bit masks to be of @@ -301,19 +302,16 @@ bool NormObject::HandleInfoRequest(bool holdoff) bool increasedRepair = false; if (info_ptr || (NormSession::FTI_INFO == session.SenderFtiMode())) { - if (!repair_info) + if (!repair_info && !pending_info) { increasedRepair = true; if (holdoff) { - if (pending_info) - increasedRepair = false; - else - pending_info = true; + pending_info = true; + pending_info_repair = true; } else { - pending_info = true; // does this really need to be done? repair_info = true; } } @@ -392,7 +390,9 @@ bool NormObject::TxReset(NormBlockId firstBlock, bool requeue) { increasedRepair = true; pending_info = true; + pending_info_repair = !requeue; } + if (requeue) pending_info_repair = false; repair_info = false; repair_mask.Reset(firstBlock.GetValue()); repair_mask.Xor(pending_mask); @@ -415,7 +415,12 @@ bool NormObject::TxReset(NormBlockId firstBlock, bool requeue) nparity, session.SenderAutoParity(), segment_size); - if (requeue) block->ClearFlag(NormBlock::IN_REPAIR); // since we're requeuing + if (requeue) + { + block->ClearFlag(NormBlock::IN_REPAIR); // since we're requeuing + block->TxResetOriginalPass(GetBlockSize(blockId), + session.SenderAutoParity()); + } } } if (requeue) @@ -472,10 +477,11 @@ bool NormObject::TxUpdateBlock(NormBlock* theBlock, bool NormObject::ActivateRepairs() { bool repairsActivated = false; - // Activate repair of info if applicable (TBD - how to flag info message as repair???) + // Activate repair of info if applicable. if (repair_info) { pending_info = true; + pending_info_repair = true; repair_info = false; repairsActivated = true; } @@ -1857,10 +1863,12 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) } if (pending_info) { - // (TBD) set REPAIR_FLAG for retransmitted info + if (pending_info_repair) + msg->SetFlag(NormObjectMsg::FLAG_REPAIR); NormInfoMsg* infoMsg = static_cast(msg); infoMsg->SetInfo(info_ptr, info_len); pending_info = false; + pending_info_repair = false; return true; } // This block gets the next pending block/segment @@ -1930,7 +1938,10 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) block->TxInit(blockId, numData, session.SenderAutoParity()); //if (blockId < max_pending_block) if (Compare(blockId, max_pending_block) < 0) + { block->SetFlag(NormBlock::IN_REPAIR); + block->TxCompleteOriginalPass(); + } while (!block_buffer.Insert(block)) { //ASSERT(STREAM == type); @@ -2013,7 +2024,8 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) { PLOG(PL_ERROR, "NormObject::NextSenderMsg() node>%lu Warning! can't repair old stream segment\n", (unsigned long)LocalNodeId()); - block->UnsetPending(segmentId); + block->UnsetPending(segmentId); + block->TxAdvanceOriginal(segmentId); if (!block->IsPending()) { // End of old block reached @@ -2021,7 +2033,10 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) pending_mask.Unset(blockId.GetValue()); // for EMCON sending, mark NORM_INFO for re-transmission, if applicable if (session.SndrEmcon() && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode()))) + { + if (!pending_info) pending_info_repair = false; pending_info = true; + } } block = NULL; continue; //return NextSenderMsg(msg); @@ -2072,9 +2087,11 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) #endif // if/else SIMULATE } } // end while (NULL == block) - block->UnsetPending(segmentId); - //if (block->InRepair()) - // data->SetFlag(NormObjectMsg::FLAG_REPAIR); + const bool originalTransmission = block->TxIsOriginalPending(segmentId); + block->UnsetPending(segmentId); + block->TxAdvanceOriginal(segmentId); + if (block->InRepair() && !originalTransmission) + data->SetFlag(NormObjectMsg::FLAG_REPAIR); data->SetFecPayloadId(fec_id, blockId.GetValue(), segmentId, numData, fec_m); if (!block->IsPending()) { @@ -2083,7 +2100,10 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) pending_mask.Unset(blockId.GetValue()); // for EMCON sending, mark NORM_INFO for re-transmission, if applicable if (session.SndrEmcon() && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode()))) + { + if (!pending_info) pending_info_repair = false; pending_info = true; + } // Advance sender use of "max_pending_block" so we always // know when a block should be flagged as IN_REPAIR if (blockId == max_pending_block) diff --git a/src/common/normSegment.cpp b/src/common/normSegment.cpp index e2a2fb7..faeffd4 100755 --- a/src/common/normSegment.cpp +++ b/src/common/normSegment.cpp @@ -90,7 +90,8 @@ char* NormSegmentPool::Get() // NormBlock Implementation NormBlock::NormBlock() - : size(0), segment_table(NULL), erasure_count(0), parity_count(0), next(NULL) + : size(0), segment_table(NULL), erasure_count(0), parity_count(0), + original_tx_next(0), original_tx_end(0), next(NULL) { } @@ -125,6 +126,7 @@ bool NormBlock::Init(UINT16 totalSize) parity_count = 0; parity_offset = 0; seg_size_max = 0; + original_tx_next = original_tx_end = 0; return true; } // end NormBlock::Init() @@ -144,6 +146,7 @@ void NormBlock::Destroy() segment_table = (char**)NULL; } erasure_count = parity_count = size = 0; + original_tx_next = original_tx_end = 0; } // end NormBlock::Destroy() void NormBlock::EmptyToPool(NormSegmentPool& segmentPool) @@ -335,6 +338,7 @@ bool NormBlock::TxUpdate(NormSegmentId nextId, NormSegmentId lastId, } } } + if (increasedRepair) SetFlag(IN_REPAIR); return increasedRepair; } // end NormBlock::TxUpdate() diff --git a/tests/normRepairFlagTest.cpp b/tests/normRepairFlagTest.cpp new file mode 100644 index 0000000..05b31b4 --- /dev/null +++ b/tests/normRepairFlagTest.cpp @@ -0,0 +1,228 @@ +#include "normMessage.h" +#include "normNode.h" +#include "normObject.h" +#include "normSession.h" +#include "protoDispatcher.h" + +#include +#include +#include + +#define TEST_ASSERT(condition) \ + do \ + { \ + if (!(condition)) \ + { \ + fprintf(stderr, "TEST FAILED: %s (line %d)\n", #condition, __LINE__); \ + exit(1); \ + } \ + } while (0) + +static void TestBlockRepairState() +{ + NormBlock block; + TEST_ASSERT(block.Init(8)); + NormBlockId blockId(0); + + block.TxInit(blockId, 4, 0); + TEST_ASSERT(!block.InRepair()); + TEST_ASSERT(block.TxIsOriginalPending(0)); + block.TxAdvanceOriginal(0); + TEST_ASSERT(!block.TxIsOriginalPending(0)); + TEST_ASSERT(block.TxIsOriginalPending(1)); + block.ClearPending(); + TEST_ASSERT(block.TxUpdate(0, 0, 4, 4, 1)); + TEST_ASSERT(block.InRepair()); + + block.TxInit(blockId, 4, 0); + block.ClearPending(); + TEST_ASSERT(block.HandleSegmentRequest(0, 0, 4, 4, 1)); + TEST_ASSERT(!block.InRepair()); + TEST_ASSERT(block.ActivateRepairs(4)); + TEST_ASSERT(block.InRepair()); + + block.TxRecover(blockId, 4, 4); + TEST_ASSERT(block.InRepair()); + TEST_ASSERT(!block.TxIsOriginalPending(0)); +} + +static void TestSyncPolicy(NormSession& session) +{ + NormSenderNode sender(session, 2); + NormDataMsg msg; + msg.Init(); + msg.SetFecId(129); + msg.SetFecPayloadId(129, 0, 0, 1, 8); + + sender.SetSyncPolicy(NormSenderNode::SYNC_CURRENT); + TEST_ASSERT(sender.SyncTest(msg)); + msg.SetFlag(NormObjectMsg::FLAG_REPAIR); + TEST_ASSERT(!sender.SyncTest(msg)); + sender.SetSyncPolicy(NormSenderNode::SYNC_ALL); + TEST_ASSERT(sender.SyncTest(msg)); +} + +static void TestObjectMessageFlags(NormSession& session) +{ + char payload[64]; + memset(payload, 0xa5, sizeof(payload)); + + { + const char info[] = "metadata"; + NormDataObject object(session, NULL, NormObjectId(1), NULL); + TEST_ASSERT(object.Open(payload, sizeof(payload), false, info, sizeof(info))); + + NormDataMsg msg; + // An already pending original INFO satisfies a repair request without + // being reclassified as reactive repair. + TEST_ASSERT(!object.HandleInfoRequest(false)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::INFO == msg.GetType()); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + + TEST_ASSERT(object.HandleInfoRequest(false)); + TEST_ASSERT(object.ActivateRepairs()); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::INFO == msg.GetType()); + TEST_ASSERT(msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + + TEST_ASSERT(object.HandleInfoRequest(true)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::INFO == msg.GetType()); + TEST_ASSERT(msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + + TEST_ASSERT(object.TxReset(NormBlockId(0), true)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::INFO == msg.GetType()); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + } + + { + session.SenderSetAutoParity(1); + NormDataObject object(session, NULL, NormObjectId(2), NULL); + TEST_ASSERT(object.Open(payload, sizeof(payload), false)); + + NormDataMsg msg; + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + + NormBlock* block = object.FindBlock(NormBlockId(0)); + TEST_ASSERT(NULL != block); + TEST_ASSERT(object.TxUpdateBlock(block, 0, 0, 1)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + + TEST_ASSERT(object.TxReset(NormBlockId(0), true)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + + TEST_ASSERT(block->HandleSegmentRequest(0, 0, 1, 4, 1)); + TEST_ASSERT(object.ActivateRepairs()); + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + session.SenderSetAutoParity(0); + } +} + +static void TestMidFirstPassRepairFlags(NormSession& session) +{ + char payload[4 * 64]; + memset(payload, 0xa5, sizeof(payload)); + NormSenderNode syncNode(session, 3); + syncNode.SetSyncPolicy(NormSenderNode::SYNC_CURRENT); + + session.SenderSetAutoParity(0); + { + NormDataObject object(session, NULL, NormObjectId(3), NULL); + TEST_ASSERT(object.Open(payload, sizeof(payload), false)); + + NormDataMsg msg; + + // Start the original pass, leaving source symbols 1 through 3 unsent. + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(NormMsg::DATA == msg.GetType()); + TEST_ASSERT(0 == msg.GetFecSymbolId(8)); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(syncNode.SyncTest(msg)); + + // Simulate a NACK arriving during the original pass. Symbol 0 is now + // pending again as reactive repair, ahead of the untouched source data. + NormBlock* block = object.FindBlock(NormBlockId(0)); + TEST_ASSERT(NULL != block); + TEST_ASSERT(object.TxUpdateBlock(block, 0, 0, 1)); + + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(0 == msg.GetFecSymbolId(8)); + TEST_ASSERT(msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(!syncNode.SyncTest(msg)); + + // The repair state must not leak onto source symbols that are still + // being transmitted for the first time. + for (UINT16 symbolId = 1; symbolId < 4; ++symbolId) + { + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(symbolId == msg.GetFecSymbolId(8)); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(syncNode.SyncTest(msg)); + } + } + + { + NormDataObject object(session, NULL, NormObjectId(4), NULL); + TEST_ASSERT(object.Open(payload, sizeof(payload), false)); + + NormDataMsg msg; + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(0 == msg.GetFecSymbolId(8)); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(syncNode.SyncTest(msg)); + + NormBlock* block = object.FindBlock(NormBlockId(0)); + TEST_ASSERT(NULL != block); + + // Request one fresh parity symbol while source symbols 1 through 3 + // remain in their original pass. + TEST_ASSERT(object.TxUpdateBlock(block, 4, 4, 1)); + for (UINT16 symbolId = 1; symbolId < 4; ++symbolId) + { + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(symbolId == msg.GetFecSymbolId(8)); + TEST_ASSERT(!msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(syncNode.SyncTest(msg)); + } + + TEST_ASSERT(object.NextSenderMsg(&msg)); + TEST_ASSERT(4 == msg.GetFecSymbolId(8)); + TEST_ASSERT(msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR)); + TEST_ASSERT(!syncNode.SyncTest(msg)); + } +} + +int main(int, char**) +{ + ProtoDispatcher dispatcher; + NormSessionMgr manager(dispatcher, dispatcher, &dispatcher); + NormSession* session = manager.NewSession("224.1.2.3", 0, 1); + TEST_ASSERT(NULL != session); + TEST_ASSERT(session->StartSender(1, 1024 * 1024, 64, 4, 4, 129)); + + TestBlockRepairState(); + TestSyncPolicy(*session); + TestObjectMessageFlags(*session); + TestMidFirstPassRepairFlags(*session); + + session->StopSender(); + manager.DeleteSession(session); + printf("normRepairFlagTest: PASSED\n"); + return 0; +}