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

tweaks to normCast example

pull/9/head
bebopagogo 2019-12-27 20:48:54 -05:00
parent e4a3d74512
commit 24199cd310
1 changed files with 16 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
examples/normCast.cpp
View File

@ -9,7 +9,7 @@
#include <sys/select.h> #include <sys/select.h>
#endif // if/else WIN32/UNIX #endif // if/else WIN32/UNIX
#define SHOOT_NOW 1 #define SHOOT_FIRST 1
// I usually avoid using Protlib stuff for NORM API examples to keep things clearer, // I usually avoid using Protlib stuff for NORM API examples to keep things clearer,
// but the couple of classes here are useful helpers from the Protolib C++ toolkit. // but the couple of classes here are useful helpers from the Protolib C++ toolkit.
@ -190,7 +190,6 @@ bool NormCaster::OpenNormSession(NormInstanceHandle instance, const char* addr,
} }
// Set some default parameters (maybe we should put all parameter setting in Start()) // Set some default parameters (maybe we should put all parameter setting in Start())
fprintf(stderr, "setting rx cache limit to %u\n", norm_tx_queue_max);
if (norm_tx_queue_max > 65535/2) norm_tx_queue_max = 65535/2; if (norm_tx_queue_max > 65535/2) norm_tx_queue_max = 65535/2;
NormSetRxCacheLimit(norm_session, norm_tx_queue_max); NormSetRxCacheLimit(norm_session, norm_tx_queue_max);
NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL); NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);
@ -259,7 +258,7 @@ bool NormCaster::Start(bool sender, bool receiver)
// Note: NormPreallocateRemoteSender() MUST be called AFTER NormStartReceiver() // Note: NormPreallocateRemoteSender() MUST be called AFTER NormStartReceiver()
NormPreallocateRemoteSender(norm_session, bufferSize, segmentSize, blockSize, numParity, bufferSize); NormPreallocateRemoteSender(norm_session, bufferSize, segmentSize, blockSize, numParity, bufferSize);
NormSetRxSocketBuffer(norm_session, rxSockBufferSize); NormSetRxSocketBuffer(norm_session, rxSockBufferSize);
fprintf(stderr, "normCast: receiver ready.\n"); fprintf(stderr, "normCast: receiver ready ...\n");
} }
if (sender) if (sender)
{ {
@ -357,21 +356,16 @@ void NormCaster::SendFiles()
{ {
sent_count++; sent_count++;
// Get next file name from our "tx_file_list" // Get next file name from our "tx_file_list"
if (!!StageNextTxFile()) if (!StageNextTxFile())
{ {
// We have reach end of tx_file_list, so either // We have reach end of tx_file_list, so either
// we're done or we reset tx_file_iterator // we're done or we reset tx_file_iterator
// If we're done and requesting ACK, finish up nicely with final waterrmark // If we're done and requesting ACK, finish up nicely with final waterrmark
if (norm_acking) if (norm_acking)
{ {
if (NORM_OBJECT_INVALID == norm_last_object) if ((NORM_OBJECT_INVALID != norm_last_object) && !norm_flushing)
{
TRACE("normCast sender STOP 2 ...\n");
is_running = false; // everything already sent and acked
}
else if (!norm_flushing)
{ {
// End-of-transmission ack request (not needed if "norm_flushing"
NormSetWatermark(norm_session, norm_last_object, true); NormSetWatermark(norm_session, norm_last_object, true);
} }
} }
@ -379,7 +373,7 @@ void NormCaster::SendFiles()
} }
else else
{ {
// locked by NORM flow control ... // blocked by NORM flow control ... will get cued by notification as described above
ASSERT(!TxReady()); ASSERT(!TxReady());
break; break;
} }
@ -426,7 +420,7 @@ bool NormCaster::EnqueueFileObject()
norm_tx_vacancy = false; norm_tx_vacancy = false;
return false; return false;
} }
fprintf(stderr, "normCast: enqueued \"%s\" for transmision ...\n", namePtr); fprintf(stderr, "normCast: enqueued \"%s\" for transmission ...\n", namePtr);
if (norm_acking) if (norm_acking)
{ {
// ack-based flow control has been enabled // ack-based flow control has been enabled
@ -437,7 +431,7 @@ bool NormCaster::EnqueueFileObject()
norm_last_object = object; norm_last_object = object;
norm_flow_control_pending = true; norm_flow_control_pending = true;
} }
else if (norm_flushing) // per-message acking else if (norm_flushing) // per-file flushing/acking
{ {
#ifdef SHOOT_FIRST #ifdef SHOOT_FIRST
NormSetWatermark(norm_session, object, true); NormSetWatermark(norm_session, object, true);
@ -458,7 +452,6 @@ bool NormCaster::EnqueueFileObject()
{ {
norm_last_object = object; norm_last_object = object;
} }
TRACE("norm_last_object: %p\n", norm_last_object);
} }
return true; return true;
} // end NormCaster::EnqueueFileObject() } // end NormCaster::EnqueueFileObject()
@ -480,8 +473,8 @@ void NormCaster::HandleNormEvent(const NormEvent& event)
case NORM_TX_WATERMARK_COMPLETED: case NORM_TX_WATERMARK_COMPLETED:
if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session))
{ {
//fprintf(stderr, "WATERMARK COMPLETED\n");
norm_last_object = NORM_OBJECT_INVALID; norm_last_object = NORM_OBJECT_INVALID;
bool txFilePending = TxFilePending(); // need to check this _before_ possible call to SendFiles()
if (norm_flow_control_pending) if (norm_flow_control_pending)
{ {
norm_tx_queue_count -= (norm_tx_queue_max / 2); norm_tx_queue_count -= (norm_tx_queue_max / 2);
@ -492,10 +485,12 @@ void NormCaster::HandleNormEvent(const NormEvent& event)
norm_last_object = norm_flush_object; norm_last_object = norm_flush_object;
norm_flush_object = NORM_OBJECT_INVALID; norm_flush_object = NORM_OBJECT_INVALID;
} }
if (txFilePending) SendFiles();
} }
if (NORM_OBJECT_INVALID == norm_last_object) if (!txFilePending)
{ {
TRACE("normCast sender STOP 1 ...\n"); // No more files to send and all have been acknowledged
fprintf(stderr, "normCast: final file acknowledged, exiting ...\n");
is_running = false; is_running = false;
} }
} }
@ -560,7 +555,6 @@ void NormCaster::HandleNormEvent(const NormEvent& event)
if ('/' == fileName[i]) if ('/' == fileName[i])
fileName[i] = PROTO_PATH_DELIMITER; fileName[i] = PROTO_PATH_DELIMITER;
} }
TRACE("got info renaming to: %s\n", fileName);
if (!NormFileRename(event.object, fileName)) if (!NormFileRename(event.object, fileName))
perror("normCast: rx file rename error"); perror("normCast: rx file rename error");
break; break;
@ -1011,16 +1005,13 @@ int main(int argc, char* argv[])
NormCloseDebugLog(normInstance); NormCloseDebugLog(normInstance);
fprintf(stderr, "destroying session ...\n");
normCast.CloseNormSession(); normCast.CloseNormSession();
fprintf(stderr, "destroying instance ...\n");
NormDestroyInstance(normInstance); NormDestroyInstance(normInstance);
normCast.Destroy(); normCast.Destroy();
fprintf(stderr, "normCast exiting ...\n"); fprintf(stderr, "normCast done.\n");
} // end main() } // end main()