diff --git a/doc/NormDeveloperGuide.xml b/doc/NormDeveloperGuide.xml
index 4408de52..92024083 100755
--- a/doc/NormDeveloperGuide.xml
+++ b/doc/NormDeveloperGuide.xml
@@ -885,9 +885,11 @@
enum NormSyncPolicy
{
- NORM_SYNC_CURRENT,
- NORM_SYNC_ALL
-};
+ NORM_SYNC_CURRENT,
+ NORM_SYNC_STREAM,
+ NORM_SYNC_ALL,
+ NORM_SYNC_REPAIR
+};
The use and interpretation of these values is given in the
descriptions of the NormSetDefaultSyncPolicy<
is the behavior observed by the receiver with regards to what
objects it attempts to reliably receive (via transmissions of
Negative Acknowledgements to the sender(s) or group as needed).
- There are currently two synchronization policy types defined:
+ There are currently four synchronization policy types defined:
@@ -5298,6 +5300,14 @@ void NormSetDefaultSyncPolicy<
only. (default)
+
+ NORM_SYNC_STREAM
+
+ Synchronize to the current stream while
+ attempting recovery from the beginning of its retained
+ receive window.
+
+
NORM_SYNC_ALL
@@ -5305,6 +5315,14 @@ void NormSetDefaultSyncPolicy<
objects held in sender transmit object cache and newer
objects.
+
+
+ NORM_SYNC_REPAIR
+
+ Attempt reception of the observed current object
+ and newer objects, allowing repair INFO or first-block DATA
+ to establish the initial synchronization point.
+
@@ -5327,6 +5345,16 @@ void NormSetDefaultSyncPolicy<
the NORM_SYNC_CURRENT policy is typically
recommended.
+ The NORM_SYNC_REPAIR policy uses the same
+ current-object receive window as
+ NORM_SYNC_CURRENT, but permits repair
+ NORM_INFO or repair NORM_DATA
+ from the first FEC block to establish synchronization. It is useful
+ for retained-object and rateless transfers where a receiver should
+ recover the object already in progress without requesting older
+ objects from the sender cache. FTI may arrive in INFO or DATA; this
+ policy does not require in-band FTI.
+
The NORM_SYNC_ALL policy allows newly
joining receivers much more aggressive behavior as they will
immediately NACK for all objects from the "current" object backwards
diff --git a/include/normApi.h b/include/normApi.h
index 0caa5c34..7af37c9c 100755
--- a/include/normApi.h
+++ b/include/normApi.h
@@ -146,7 +146,8 @@ typedef enum NormSyncPolicy
{
NORM_SYNC_CURRENT, // attempt to receiver current/new objects only, join mid-stream
NORM_SYNC_STREAM, // sync to current stream, but to beginning of stream
- NORM_SYNC_ALL // attempt to receive old and new objects
+ NORM_SYNC_ALL, // attempt to receive old and new objects
+ NORM_SYNC_REPAIR // accept current-object repair traffic as an initial sync point
} NORM_API_LINKAGE NormSyncPolicy;
typedef enum NormRepairBoundary
diff --git a/include/normNode.h b/include/normNode.h
index 64434db2..ee41422b 100755
--- a/include/normNode.h
+++ b/include/normNode.h
@@ -317,7 +317,8 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
{
SYNC_CURRENT, // sync to detect transmit point, iff NORM_DATA from first FEC block unless stream
SYNC_STREAM, // same as SYNC_CURRENT, but attempts to recover stream block zero
- SYNC_ALL // permiscuously sync as far back as possible given rx cache size
+ SYNC_ALL, // promiscuously sync as far back as possible given rx cache size
+ SYNC_REPAIR // sync current object from original or repair traffic, without historical rewind
};
NormSenderNode(class NormSession& theSession, NormNodeId nodeId);
diff --git a/src/common/normApp.cpp b/src/common/normApp.cpp
index 283d0151..b2aa5a90 100755
--- a/src/common/normApp.cpp
+++ b/src/common/normApp.cpp
@@ -207,6 +207,7 @@ class NormApp : public NormController, public ProtoApp
bool process_aborted_files;
bool preallocate_sender;
NormSenderNode::RepairBoundary repair_boundary;
+ NormSenderNode::SyncPolicy sync_policy;
// Debug parameters
bool tracing;
@@ -248,7 +249,8 @@ NormApp::NormApp()
acking_flushes(false), watermark_pending(false), rx_buffer_size(1024*1024), rx_sock_buffer_size(0),
rx_cache_path(NULL), post_processor(NULL), unicast_nacks(false), silent_receiver(false),
low_delay(false), realtime(false), rx_robust_factor(NormSession::DEFAULT_ROBUST_FACTOR), rx_persistent(true), process_aborted_files(false),
- preallocate_sender(false), repair_boundary(NormSenderNode::BLOCK_BOUNDARY), tracing(false), tx_loss(0.0), rx_loss(0.0)
+ preallocate_sender(false), repair_boundary(NormSenderNode::BLOCK_BOUNDARY),
+ sync_policy(NormSenderNode::SYNC_CURRENT), tracing(false), tx_loss(0.0), rx_loss(0.0)
{
control_pipe.SetListener(this, &NormApp::OnControlEvent);
control_pipe.SetNotifier(&GetSocketNotifier());
@@ -343,6 +345,7 @@ const char* const NormApp::cmd_list[] =
"+moutput", // recv message stream output
"+sendfile", // file/directory list to transmit
"+info", // 'on' | 'off' to enable/disable file info (name) transmission (default = 'on')
+ "+sync", // receiver synchronization policy: 'current', 'stream', 'repair', or 'all'
"+interval", // delay time (sec) between files (0.0 sec default)
"+repeatcount", // How many times to repeat the file/directory list tx
"+rinterval", // Interval (sec) between file/directory list repeats
@@ -416,6 +419,7 @@ void NormApp::ShowHelp()
" +moutput, // receiver message stream output\n"
" +sendfile, // file/directory list to transmit\n"
" +info, // 'on' | 'off' to enable/disable file info (name) transmission (default = 'on')\n"
+ " +sync, // receiver synchronization policy: 'current', 'stream', 'repair', or 'all'\n"
" +interval, // delay time (sec) between files (0.0 sec default)\n"
" +repeatcount, // How many times to repeat the file/directory list tx\n"
" +rinterval, // Interval (sec) between file/directory list repeats\n"
@@ -883,6 +887,23 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
return false;
}
}
+ else if (!strncmp("sync", cmd, len))
+ {
+ if (!strcmp("current", val))
+ sync_policy = NormSenderNode::SYNC_CURRENT;
+ else if (!strcmp("stream", val))
+ sync_policy = NormSenderNode::SYNC_STREAM;
+ else if (!strcmp("repair", val))
+ sync_policy = NormSenderNode::SYNC_REPAIR;
+ else if (!strcmp("all", val))
+ sync_policy = NormSenderNode::SYNC_ALL;
+ else
+ {
+ PLOG(PL_FATAL, "NormApp::OnCommand(sync) invalid policy!\n");
+ return false;
+ }
+ if (NULL != session) session->ReceiverSetDefaultSyncPolicy(sync_policy);
+ }
else if (!strncmp("interval", cmd, len))
{
if (1 != sscanf(val, "%lf", &tx_object_interval))
@@ -2383,6 +2404,7 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
session->RcvrSetRealtime(true);
session->SetRxRobustFactor(rx_robust_factor);
session->ReceiverSetDefaultRepairBoundary(repair_boundary);
+ session->ReceiverSetDefaultSyncPolicy(sync_policy);
if (!session->StartReceiver(rx_buffer_size))
{
PLOG(PL_FATAL, "NormApp::OnStartup() start receiver error!\n");
diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp
index 63619f0c..f66deed8 100755
--- a/src/common/normNode.cpp
+++ b/src/common/normNode.cpp
@@ -1782,7 +1782,8 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
if (presetStream) preset_stream = NULL; // we're using it up
// This initial "StreamUpdateStatus()" syncs the stream according to our sync policy
NormStreamObject* stream = static_cast(obj);
- if (SYNC_CURRENT == sync_policy)
+ if ((SYNC_CURRENT == sync_policy) ||
+ (SYNC_REPAIR == sync_policy))
{
// Just "sync" to first received blockId
stream->StreamUpdateStatus(blockId);
@@ -1901,20 +1902,27 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
bool NormSenderNode::SyncTest(const NormObjectMsg& msg) const
{
+ // SYNC_CURRENT and SYNC_REPAIR both establish the observed object as the
+ // receiver's current point. Restrict DATA synchronization to block zero
+ // so that a later repair block cannot establish a partial object.
+ bool currentCandidate = msg.FlagIsSet(NormObjectMsg::FLAG_STREAM) ||
+ (NormMsg::INFO == msg.GetType());
+ if (!currentCandidate && (NormMsg::DATA == msg.GetType()))
+ {
+ const NormDataMsg& data = static_cast(msg);
+ currentCandidate =
+ (NormBlockId(0) == data.GetFecBlockId(fti_data.GetFecFieldSize()));
+ }
+
switch (sync_policy)
{
case SYNC_CURRENT: // default, more conservative "sync policy"
case SYNC_STREAM:
- {
- // Allow sync on stream at any time
- bool result = msg.FlagIsSet(NormObjectMsg::FLAG_STREAM);
- // Allow sync on INFO or block zero DATA message
- result = result || (NormMsg::INFO == msg.GetType()) ?
- true : (NormBlockId(0) == ((const NormDataMsg&)msg).GetFecBlockId(fti_data.GetFecFieldSize()));
- // Never sync on repair messages
- result = result && !msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR);
- return result;
- }
+ return currentCandidate &&
+ !msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR);
+
+ case SYNC_REPAIR:
+ return currentCandidate;
case SYNC_ALL: // sync on anything
return true;
@@ -2010,6 +2018,7 @@ void NormSenderNode::Sync(NormObjectId objectId)
{
case SYNC_CURRENT: // this is the usual default
case SYNC_STREAM:
+ case SYNC_REPAIR:
sync_id = next_id = max_pending_object = objectId;
break;
case SYNC_ALL: // gratuitously sync for anything in our "range"
diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp
index 24299aa8..4bfc39ae 100755
--- a/src/common/normObject.cpp
+++ b/src/common/normObject.cpp
@@ -3119,6 +3119,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId)
read_offset = 0;
sender->DecrementResyncCount(); // correction since stream sync here will falsely increment
if ((NormSenderNode::SYNC_CURRENT != sender->GetSyncPolicy()) &&
+ (NormSenderNode::SYNC_REPAIR != sender->GetSyncPolicy()) &&
(0 != blockId.GetValue()))
{
stream_broken = true;
diff --git a/src/dotnet/design/Mil/Navy/Nrl/Norm/Enums/NormSyncPolicy.puml b/src/dotnet/design/Mil/Navy/Nrl/Norm/Enums/NormSyncPolicy.puml
index 870db564..12a76f27 100644
--- a/src/dotnet/design/Mil/Navy/Nrl/Norm/Enums/NormSyncPolicy.puml
+++ b/src/dotnet/design/Mil/Navy/Nrl/Norm/Enums/NormSyncPolicy.puml
@@ -3,5 +3,6 @@ enum NormSyncPolicy {
NORM_SYNC_CURRENT,
NORM_SYNC_STREAM,
NORM_SYNC_ALL,
+ NORM_SYNC_REPAIR,
}
@enduml
diff --git a/src/dotnet/src/Mil.Navy.Nrl.Norm/Enums/NormSyncPolicy.cs b/src/dotnet/src/Mil.Navy.Nrl.Norm/Enums/NormSyncPolicy.cs
index deb54b54..bba33a1a 100644
--- a/src/dotnet/src/Mil.Navy.Nrl.Norm/Enums/NormSyncPolicy.cs
+++ b/src/dotnet/src/Mil.Navy.Nrl.Norm/Enums/NormSyncPolicy.cs
@@ -17,6 +17,11 @@ public enum NormSyncPolicy
/// Attempt recovery and reliable reception of all objects
/// held in sender transmit object cache and newer objects.
///
- NORM_SYNC_ALL
+ NORM_SYNC_ALL,
+ ///
+ /// Accept repair traffic for the observed current object without
+ /// requesting older objects from the sender cache.
+ ///
+ NORM_SYNC_REPAIR
}
}
diff --git a/src/dotnet/tests/Mil.Navy.Nrl.Norm.IntegrationTests/NormSessionTests.cs b/src/dotnet/tests/Mil.Navy.Nrl.Norm.IntegrationTests/NormSessionTests.cs
index 39d24d5c..9c4c5708 100644
--- a/src/dotnet/tests/Mil.Navy.Nrl.Norm.IntegrationTests/NormSessionTests.cs
+++ b/src/dotnet/tests/Mil.Navy.Nrl.Norm.IntegrationTests/NormSessionTests.cs
@@ -1915,6 +1915,13 @@ public void SetsDefaultSyncPolicy_NORM_SYNC_ALL()
_normSession.SetDefaultSyncPolicy(syncPolicy);
}
+ [Fact]
+ public void SetsDefaultSyncPolicy_NORM_SYNC_REPAIR()
+ {
+ var syncPolicy = NormSyncPolicy.NORM_SYNC_REPAIR;
+ _normSession.SetDefaultSyncPolicy(syncPolicy);
+ }
+
[Fact]
public void SetsDefaultNackingMode_NORM_NACK_NONE()
{
@@ -2807,4 +2814,4 @@ public void NormStreamSetsAutoFlush_FLUSH_ACTIVE()
}
}
}
-}
\ No newline at end of file
+}
diff --git a/src/java/src/mil/navy/nrl/norm/enums/NormSyncPolicy.java b/src/java/src/mil/navy/nrl/norm/enums/NormSyncPolicy.java
index 52374568..88ea7345 100755
--- a/src/java/src/mil/navy/nrl/norm/enums/NormSyncPolicy.java
+++ b/src/java/src/mil/navy/nrl/norm/enums/NormSyncPolicy.java
@@ -12,5 +12,6 @@
public enum NormSyncPolicy {
NORM_SYNC_CURRENT,
NORM_SYNC_STREAM,
- NORM_SYNC_ALL;
+ NORM_SYNC_ALL,
+ NORM_SYNC_REPAIR;
}
diff --git a/src/pynorm/constants.py b/src/pynorm/constants.py
index c7e4d0aa..09078b4a 100755
--- a/src/pynorm/constants.py
+++ b/src/pynorm/constants.py
@@ -75,7 +75,8 @@ class SyncPolicy(enum.Enum):
'''
CURRENT = 0
STREAM = 1
- ALL = 2
+ ALL = 2
+ REPAIR = 3
class TrackingStatus(enum.Enum): #enum NormTrackingStatus
NONE = 0
diff --git a/src/rust/norm/src/types.rs b/src/rust/norm/src/types.rs
index b19f1edd..1f09e41f 100644
--- a/src/rust/norm/src/types.rs
+++ b/src/rust/norm/src/types.rs
@@ -242,6 +242,8 @@ pub enum SyncPolicy {
Stream = NormSyncPolicy_NORM_SYNC_STREAM as u32,
/// Sync to all data, old and new
All = NormSyncPolicy_NORM_SYNC_ALL as u32,
+ /// Accept repair traffic for the observed current object only
+ Repair = NormSyncPolicy_NORM_SYNC_REPAIR as u32,
}
impl From for SyncPolicy {
@@ -250,6 +252,7 @@ impl From for SyncPolicy {
NormSyncPolicy_NORM_SYNC_CURRENT => SyncPolicy::Current,
NormSyncPolicy_NORM_SYNC_STREAM => SyncPolicy::Stream,
NormSyncPolicy_NORM_SYNC_ALL => SyncPolicy::All,
+ NormSyncPolicy_NORM_SYNC_REPAIR => SyncPolicy::Repair,
_ => SyncPolicy::Current,
}
}
@@ -261,6 +264,7 @@ impl From for NormSyncPolicy {
SyncPolicy::Current => NormSyncPolicy_NORM_SYNC_CURRENT,
SyncPolicy::Stream => NormSyncPolicy_NORM_SYNC_STREAM,
SyncPolicy::All => NormSyncPolicy_NORM_SYNC_ALL,
+ SyncPolicy::Repair => NormSyncPolicy_NORM_SYNC_REPAIR,
}
}
}
@@ -522,4 +526,4 @@ impl fmt::Display for EventType {
EventType::UserTimeout => write!(f, "UserTimeout"),
}
}
-}
\ No newline at end of file
+}
diff --git a/tests/normRepairFlagTest.cpp b/tests/normRepairFlagTest.cpp
index 05b31b45..8d8bdb0d 100644
--- a/tests/normRepairFlagTest.cpp
+++ b/tests/normRepairFlagTest.cpp
@@ -58,8 +58,30 @@ static void TestSyncPolicy(NormSession& session)
TEST_ASSERT(sender.SyncTest(msg));
msg.SetFlag(NormObjectMsg::FLAG_REPAIR);
TEST_ASSERT(!sender.SyncTest(msg));
+
+ sender.SetSyncPolicy(NormSenderNode::SYNC_REPAIR);
+ TEST_ASSERT(sender.SyncTest(msg));
+
sender.SetSyncPolicy(NormSenderNode::SYNC_ALL);
TEST_ASSERT(sender.SyncTest(msg));
+
+ NormDataMsg laterBlock;
+ laterBlock.Init();
+ laterBlock.SetFecId(129);
+ laterBlock.SetFecPayloadId(129, 1, 0, 1, 8);
+ laterBlock.SetFlag(NormObjectMsg::FLAG_REPAIR);
+ sender.SetSyncPolicy(NormSenderNode::SYNC_REPAIR);
+ TEST_ASSERT(!sender.SyncTest(laterBlock));
+ sender.SetSyncPolicy(NormSenderNode::SYNC_ALL);
+ TEST_ASSERT(sender.SyncTest(laterBlock));
+
+ NormInfoMsg repairInfo;
+ repairInfo.Init();
+ repairInfo.SetFlag(NormObjectMsg::FLAG_REPAIR);
+ sender.SetSyncPolicy(NormSenderNode::SYNC_CURRENT);
+ TEST_ASSERT(!sender.SyncTest(repairInfo));
+ sender.SetSyncPolicy(NormSenderNode::SYNC_REPAIR);
+ TEST_ASSERT(sender.SyncTest(repairInfo));
}
static void TestObjectMessageFlags(NormSession& session)