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

v1.2b3

pull/2/head
Jeff Weston 2019-09-11 11:41:23 -04:00
parent 7fdb889444
commit a5e5d140e3
11 changed files with 205 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
common/normApi.cpp
View File

@ -1018,14 +1018,28 @@ bool NormReadStream(NormObjectHandle streamHandle,
NormStreamObject* stream = NormStreamObject* stream =
static_cast<NormStreamObject*>((NormObject*)streamHandle); static_cast<NormStreamObject*>((NormObject*)streamHandle);
if (stream) if (stream)
{
result = stream->Read(buffer, numBytes); result = stream->Read(buffer, numBytes);
}
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return result; return result;
} // end NormReadStream() } // end NormReadStream()
bool NormFindStreamMsgStart(NormObjectHandle streamHandle)
{
bool result = false;
NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormStreamObject* stream =
static_cast<NormStreamObject*>((NormObject*)streamHandle);
unsigned int numBytes = 0;
if (stream)
result = stream->Read(NULL, &numBytes, true);
instance->dispatcher.ResumeThread();
}
return result;
} // end NormFindStreamMsgStart()
NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle,
const char* fileName, const char* fileName,
const char* infoPtr, const char* infoPtr,

2
common/normApi.h
View File

@ -186,6 +186,8 @@ bool NormReadStream(NormObjectHandle streamHandle,
char* buffer, char* buffer,
unsigned int* numBytes); unsigned int* numBytes);
bool NormFindStreamMsgStart(NormObjectHandle streamHandle);
// NormFileObject Functions // NormFileObject Functions
NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle,
const char* fileName, const char* fileName,

21
common/normApp.cpp
View File

@ -108,6 +108,7 @@ class NormApp : public NormController, public ProtoApp
// NormSession client-only parameters // NormSession client-only parameters
unsigned long rx_buffer_size; // bytes unsigned long rx_buffer_size; // bytes
unsigned int rx_sock_buffer_size;
NormFileList rx_file_cache; NormFileList rx_file_cache;
char* rx_cache_path; char* rx_cache_path;
NormPostProcessor* post_processor; NormPostProcessor* post_processor;
@ -129,7 +130,7 @@ NormApp::NormApp()
push_mode(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE), push_mode(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE),
input_messaging(false), input_msg_length(0), input_msg_index(0), input_messaging(false), input_msg_length(0), input_msg_index(0),
output_index(0), output_messaging(false), output_msg_length(0), output_msg_sync(false), output_index(0), output_messaging(false), output_msg_length(0), output_msg_sync(false),
address(NULL), port(0), ttl(3), loopback(false), interface_name(NULL), address(NULL), port(0), ttl(32), loopback(false), interface_name(NULL),
tx_rate(64000.0), cc_enable(false), tx_rate(64000.0), cc_enable(false),
segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0), segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0),
backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR), backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR),
@ -137,7 +138,8 @@ NormApp::NormApp()
group_size(NormSession::DEFAULT_GSIZE_ESTIMATE), group_size(NormSession::DEFAULT_GSIZE_ESTIMATE),
tx_buffer_size(1024*1024), tx_buffer_size(1024*1024),
tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true), tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true),
rx_buffer_size(1024*1024), rx_cache_path(NULL), unicast_nacks(false), silent_client(false), rx_buffer_size(1024*1024), rx_sock_buffer_size(0),
rx_cache_path(NULL), unicast_nacks(false), silent_client(false),
tracing(false), tx_loss(0.0), rx_loss(0.0) tracing(false), tx_loss(0.0), rx_loss(0.0)
{ {
@ -215,6 +217,7 @@ const char* const NormApp::cmd_list[] =
"+gsize", // Set sender's group size estimate "+gsize", // Set sender's group size estimate
"+txbuffer", // Size of sender's buffer "+txbuffer", // Size of sender's buffer
"+rxbuffer", // Size receiver allocates for buffering each sender "+rxbuffer", // Size receiver allocates for buffering each sender
"+rxsockbuffer", // Optional recv socket buffer size.
"-unicastNacks", // unicast instead of multicast feedback messages "-unicastNacks", // unicast instead of multicast feedback messages
"-silentClient", // "silent" (non-nacking) client (EMCON mode) "-silentClient", // "silent" (non-nacking) client (EMCON mode)
"+processor", // receive file post processing command "+processor", // receive file post processing command
@ -687,6 +690,16 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
return false; return false;
} }
} }
else if (!strncmp("rxsockbuffer", cmd, len))
{
if (1 != sscanf(val, "%u", &rx_sock_buffer_size))
{
DMSG(0, "NormApp::OnCommand(rxsockbuffer) invalid value!\n");
return false;
}
if (session && (rx_sock_buffer_size > 0))
session->SetRxSocketBuffer(rx_sock_buffer_size);
}
else if (!strncmp("unicastNacks", cmd, len)) else if (!strncmp("unicastNacks", cmd, len))
{ {
unicast_nacks = true; unicast_nacks = true;
@ -1186,7 +1199,7 @@ void NormApp::Notify(NormController::Event event,
DMSG(0, "NormApp::Notify() detected broken stream ...\n"); DMSG(0, "NormApp::Notify() detected broken stream ...\n");
output_msg_length = output_index = 0; output_msg_length = output_index = 0;
output_msg_sync = false; output_msg_sync = false;
continue; break;
} }
} }
else else
@ -1464,6 +1477,8 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
session_mgr.Destroy(); session_mgr.Destroy();
return false; return false;
} }
if (rx_sock_buffer_size > 0)
session->SetRxSocketBuffer(rx_sock_buffer_size);
} }
return true; return true;
} }

1
common/normMessage.h
View File

@ -198,6 +198,7 @@ class NormObjectId
bool operator!=(const NormObjectId& id) const bool operator!=(const NormObjectId& id) const
{return (value != id.value);} {return (value != id.value);}
NormObjectId& operator++(int) {value++; return *this;} NormObjectId& operator++(int) {value++; return *this;}
NormObjectId& operator--(int) {value--; return *this;}
private: private:
UINT16 value; UINT16 value;

38
common/normNode.cpp
View File

@ -207,7 +207,9 @@ void NormServerNode::FreeBuffers()
while ((obj = rx_table.Find(rx_table.RangeLo()))) while ((obj = rx_table.Find(rx_table.RangeLo())))
{ {
session.Notify(NormController::RX_OBJECT_ABORTED, this, obj); session.Notify(NormController::RX_OBJECT_ABORTED, this, obj);
DeleteObject(obj, 6); DeleteObject(obj);
// We do the following to remember which objects were pending
rx_pending_mask.Set(obj->GetId());
} }
segment_pool.Destroy(); segment_pool.Destroy();
block_pool.Destroy(); block_pool.Destroy();
@ -685,7 +687,7 @@ void NormServerNode::CalculateGrttResponse(const struct timeval& currentTime,
} }
} // end NormServerNode::CalculateGrttResponse() } // end NormServerNode::CalculateGrttResponse()
void NormServerNode::DeleteObject(NormObject* obj, int which) void NormServerNode::DeleteObject(NormObject* obj)
{ {
if (rx_table.Remove(obj)) if (rx_table.Remove(obj))
rx_pending_mask.Unset(obj->GetId()); rx_pending_mask.Unset(obj->GetId());
@ -866,10 +868,16 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
} }
else else
{ {
// The hacky use of "sync_id" here keeps the debug message from
// printing too often.
if (0 == sync_id) if (0 == sync_id)
{ {
DMSG(0, "NormServerNode::HandleObjectMessage() waiting to sync ...\n"); DMSG(0, "NormServerNode::HandleObjectMessage() waiting to sync ...\n");
sync_id = 1; sync_id = 100;
}
else
{
sync_id--;
} }
return; return;
} }
@ -939,13 +947,13 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
} }
else else
{ {
DeleteObject(obj, 7); DeleteObject(obj);
obj = NULL; obj = NULL;
} }
} }
else else
{ {
DeleteObject(obj, 1); DeleteObject(obj);
obj = NULL; obj = NULL;
} }
break; break;
@ -955,7 +963,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
{ {
DMSG(0, "NormServerNode::HandleObjectMessage() node>%lu server>%lu " DMSG(0, "NormServerNode::HandleObjectMessage() node>%lu server>%lu "
"new obj>%hu - no FTI provided!\n", LocalNodeId(), GetId(), (UINT16)objectId); "new obj>%hu - no FTI provided!\n", LocalNodeId(), GetId(), (UINT16)objectId);
DeleteObject(obj, 2); DeleteObject(obj);
obj = NULL; obj = NULL;
} }
} }
@ -985,7 +993,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
{ {
// Streams never complete // Streams never complete
session.Notify(NormController::RX_OBJECT_COMPLETED, this, obj); session.Notify(NormController::RX_OBJECT_COMPLETED, this, obj);
DeleteObject(obj, 3); DeleteObject(obj);
completion_count++; completion_count++;
} }
} }
@ -1058,7 +1066,7 @@ void NormServerNode::Sync(NormObjectId objectId)
while ((obj = rx_table.Find(rx_table.RangeLo()))) while ((obj = rx_table.Find(rx_table.RangeLo())))
{ {
session.Notify(NormController::RX_OBJECT_ABORTED, this, obj); session.Notify(NormController::RX_OBJECT_ABORTED, this, obj);
DeleteObject(obj, 4); DeleteObject(obj);
failure_count++; failure_count++;
} }
rx_pending_mask.Clear(); rx_pending_mask.Clear();
@ -1070,7 +1078,7 @@ void NormServerNode::Sync(NormObjectId objectId)
(obj->GetId() < objectId)) (obj->GetId() < objectId))
{ {
session.Notify(NormController::RX_OBJECT_ABORTED, this, obj); session.Notify(NormController::RX_OBJECT_ABORTED, this, obj);
DeleteObject(obj, 5); DeleteObject(obj);
failure_count++; failure_count++;
} }
unsigned long numBits = (UINT16)(objectId - firstPending) + 1; unsigned long numBits = (UINT16)(objectId - firstPending) + 1;
@ -1583,15 +1591,15 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned
else 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);
double rttEstimate = rtt_confirmed ? rtt_estimate : grtt_estimate; double rttEstimate = rtt_confirmed ? rtt_estimate : grtt_estimate;
if (interval < rttEstimate) // We put a 0.100 sec lower bound on our rttEstimate for the recv_rate measurement
{ // interval because of the typical limited granularity of our system clock
recv_accumulator += msgSize; rttEstimate = rttEstimate < 0.1 ? 0.1 : rttEstimate;
} recv_accumulator += msgSize;
else if (interval >= rttEstimate)
{ {
recv_rate = ((double)(recv_accumulator)) / interval; recv_rate = ((double)(recv_accumulator)) / interval;
prev_update_time = currentTime; prev_update_time = currentTime;
recv_accumulator = msgSize; recv_accumulator = 0;
} }
} }
else else

2
common/normNode.h
View File

@ -281,7 +281,7 @@ class NormServerNode : public NormNode
} }
void SetPending(NormObjectId objectId); void SetPending(NormObjectId objectId);
void DeleteObject(NormObject* obj, int which); void DeleteObject(NormObject* obj);
UINT16 SegmentSize() {return segment_size;} UINT16 SegmentSize() {return segment_size;}
UINT16 BlockSize() {return ndata;} UINT16 BlockSize() {return ndata;}

21
common/normObject.cpp
View File

@ -2315,7 +2315,8 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
block = stream_buffer.Find(stream_buffer.RangeLo()); block = stream_buffer.Find(stream_buffer.RangeLo());
if (block->IsPending()) broken = true; if (block->IsPending()) broken = true;
// This loop feeds any received user data segments to the application // This loop feeds any received user data segments to the application
// (if the application doesn't want the data, it's lost) // (if the application doesn't want the data, it's lost!)
bool dataLost = false;
while (block->IsPending()) while (block->IsPending())
{ {
// Notify app for stream data salvage // Notify app for stream data salvage
@ -2330,9 +2331,8 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
if (tempOffset == read_offset) if (tempOffset == read_offset)
{ {
// App didn't want any data here, purge segment // App didn't want any data here, purge segment
DMSG(0, "NormStreamObject::WriteSegment() app didn't want data ?!, blockId:%lu\n", //ASSERT(0);
(UINT32)block->GetId()); dataLost = true;
ASSERT(0);
char* s = block->DetachSegment(read_index.segment); char* s = block->DetachSegment(read_index.segment);
segment_pool.Put(s); segment_pool.Put(s);
block->UnsetPending(read_index.segment); block->UnsetPending(read_index.segment);
@ -2345,6 +2345,9 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
break; break;
} }
} }
if (dataLost)
DMSG(0, "NormStreamObject::WriteSegment() broken stream data dropped!\n");
if (block) if (block)
{ {
// if the app didn't consume the block, we must // if the app didn't consume the block, we must
@ -2443,7 +2446,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
{ {
// DMSG(0, "NormStreamObject::Read() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty()); // DMSG(0, "NormStreamObject::Read() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty());
*buflen = bytesRead; *buflen = bytesRead;
return true; if (bytesRead > 0)
return true;
else
return findMsgStart ? false : true;
} }
char* segment = block->Segment(read_index.segment); char* segment = block->Segment(read_index.segment);
if (!segment) if (!segment)
@ -2451,7 +2457,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
//DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n", //DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n",
// (UINT32)read_index.block, read_index.segment); // (UINT32)read_index.block, read_index.segment);
*buflen = bytesRead; *buflen = bytesRead;
return true; if (bytesRead > 0)
return true;
else
return findMsgStart ? false : true;
} }
UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment);

75
common/normSession.cpp
View File

@ -1378,22 +1378,26 @@ double NormSession::CalculateRtt(const struct timeval& currentTime,
void NormSession::ServerUpdateGrttEstimate(double clientRtt) void NormSession::ServerUpdateGrttEstimate(double clientRtt)
{ {
grtt_response = true; grtt_response = true;
if ((clientRtt > grtt_current_peak) || !address.IsMulticast()) //if ((clientRtt > grtt_current_peak) || !address.IsMulticast())
if ((clientRtt > grtt_measured) || !address.IsMulticast())
{ {
// Immediately incorporate bigger RTT's // Immediately incorporate bigger RTT's
grtt_current_peak = clientRtt; grtt_current_peak = clientRtt;
if ((clientRtt > grtt_measured) || !address.IsMulticast()) //if ((clientRtt > grtt_measured) || !address.IsMulticast())
{ {
grtt_decrease_delay_count = DEFAULT_GRTT_DECREASE_DELAY; grtt_decrease_delay_count = DEFAULT_GRTT_DECREASE_DELAY;
grtt_measured = 0.25 * grtt_measured + 0.75 * clientRtt; //grtt_measured = 0.25 * grtt_measured + 0.75 * clientRtt;
grtt_measured = 0.9 * grtt_measured + 0.1 * clientRtt;
if (grtt_measured > grtt_max) grtt_measured = grtt_max; if (grtt_measured > grtt_max) grtt_measured = grtt_max;
double pktInterval = ((double)(44+segment_size))/tx_rate; double pktInterval = ((double)(44+segment_size))/tx_rate;
UINT8 grttQuantizedOld = grtt_quantized; UINT8 grttQuantizedOld = grtt_quantized;
grtt_quantized = NormQuantizeRtt(MAX(pktInterval, grtt_measured)); grtt_quantized = NormQuantizeRtt(MAX(pktInterval, grtt_measured));
// Calculate grtt_advertised since quantization rounds upward // Calculate grtt_advertised since quantization rounds upward
grtt_advertised = NormUnquantizeRtt(grtt_quantized); grtt_advertised = NormUnquantizeRtt(grtt_quantized);
double clrRtt = cc_node_list.Head() ? ((NormCCNode*)cc_node_list.Head())->GetRtt() : -1;
if (grttQuantizedOld != grtt_quantized) if (grttQuantizedOld != grtt_quantized)
DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt: %lf sec.\n", DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt>%lf sec\n",
LocalNodeId(), grtt_advertised); LocalNodeId(), grtt_advertised);
} }
} }
@ -1433,7 +1437,7 @@ void NormSession::ServerHandleCCFeedback(NormNodeId nodeId,
} }
if (!cc_enable) return; if (!cc_enable) return;
// Adjust ccRtt if we have state on this nodeId // Adjust ccRtt if we already have state on this nodeId
NormCCNode* node = (NormCCNode*)cc_node_list.FindNodeById(nodeId); NormCCNode* node = (NormCCNode*)cc_node_list.FindNodeById(nodeId);
if (node) ccRtt = node->UpdateRtt(ccRtt); if (node) ccRtt = node->UpdateRtt(ccRtt);
@ -1818,7 +1822,8 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
if ((OBJECT == requestLevel) || (INFO == requestLevel)) if ((OBJECT == requestLevel) || (INFO == requestLevel))
{ {
nextObjectId++; nextObjectId++;
if (nextObjectId > lastObjectId) inRange = false; if (nextObjectId > lastObjectId)
inRange = false;
} }
else else
{ {
@ -1842,7 +1847,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
startTimer = true; startTimer = true;
} }
} }
} } // end if (freshObject)
ASSERT(object); ASSERT(object);
switch (requestLevel) switch (requestLevel)
@ -1871,11 +1876,10 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
nextObjectId++; nextObjectId++;
if (nextObjectId > lastObjectId) inRange = false; if (nextObjectId > lastObjectId) inRange = false;
break; break;
case BLOCK: case BLOCK:
DMSG(8, "NormSession::ServerHandleNackMessage(BLOCK) obj>%hu blks>%lu:%lu\n", DMSG(8, "NormSession::ServerHandleNackMessage(BLOCK) obj>%hu blks>%lu:%lu\n",
(UINT16)nextObjectId, (UINT32)nextBlockId, (UINT32)lastBlockId); (UINT16)nextObjectId, (UINT32)nextBlockId, (UINT32)lastBlockId);
inRange = false; // BLOCK requests are processed in one pass
// (TBD) if entire object is TxReset(), continue // (TBD) if entire object is TxReset(), continue
if (object->IsStream()) if (object->IsStream())
{ {
@ -1893,7 +1897,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else if (nextObjectId < txObjectIndex) else if (nextObjectId < txObjectIndex)
{ {
attemptLock = false; // NACK arrived too late attemptLock = false; // NACK arrived too late to be useful
} }
} }
@ -1907,21 +1911,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
{ {
DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu LockBlocks() failure\n", DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu LockBlocks() failure\n",
LocalNodeId()); LocalNodeId());
inRange = false;
if (!squelchQueued) if (!squelchQueued)
{ {
ServerQueueSquelch(nextObjectId); ServerQueueSquelch(nextObjectId);
squelchQueued = true; squelchQueued = true;
} }
continue; break;
} }
} }
else else
{ {
inRange = false; break; // ignore late arriving NACK
continue;
} }
} } // end if (object->IsStream()
if (holdoff) if (holdoff)
{ {
if (nextObjectId == txObjectIndex) if (nextObjectId == txObjectIndex)
@ -1946,21 +1948,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
object->HandleBlockRequest(nextBlockId, lastBlockId); object->HandleBlockRequest(nextBlockId, lastBlockId);
startTimer = true; startTimer = true;
} }
inRange = false;
break; break;
case SEGMENT: case SEGMENT:
DMSG(8, "NormSession::ServerHandleNackMessage(SEGMENT) obj>%hu blk>%lu segs>%hu:%hu\n", DMSG(8, "NormSession::ServerHandleNackMessage(SEGMENT) obj>%hu blk>%lu segs>%hu:%hu\n",
(UINT16)nextObjectId, (UINT32)nextBlockId, (UINT16)nextObjectId, (UINT32)nextBlockId,
(UINT32)nextSegmentId, (UINT32)lastSegmentId); (UINT32)nextSegmentId, (UINT32)lastSegmentId);
inRange = false; // SEGMENT repairs are also handled in one pass
if (nextBlockId != prevBlockId) freshBlock = true; if (nextBlockId != prevBlockId) freshBlock = true;
if (freshBlock) if (freshBlock)
{ {
freshBlock = false;
// Is this entire block already repair pending? // Is this entire block already repair pending?
if (object->IsRepairSet(nextBlockId)) if (object->IsRepairSet(nextBlockId))
{ break;
inRange = false;
continue;
}
if (!(block = object->FindBlock(nextBlockId))) if (!(block = object->FindBlock(nextBlockId)))
{ {
// Is this entire block already tx pending? // Is this entire block already tx pending?
@ -1974,22 +1974,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu " DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu "
"recvd repair request for old stream block(%lu) ...\n", "recvd repair request for old stream block(%lu) ...\n",
LocalNodeId(), (UINT32)nextBlockId); LocalNodeId(), (UINT32)nextBlockId);
inRange = false;
if (!squelchQueued) if (!squelchQueued)
{ {
ServerQueueSquelch(nextObjectId); ServerQueueSquelch(nextObjectId);
squelchQueued = true; squelchQueued = true;
} }
continue;
} }
else else
{ {
// Resource constrained, move on. // Resource constrained, move on.
DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu " DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu "
"Warning - server is resource contrained ...\n"); "Warning - server is resource contrained ...\n");
inRange = false;
continue;
} }
break;
} }
} }
else else
@ -1997,10 +1994,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
// Entire block already tx pending, don't recover // Entire block already tx pending, don't recover
DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu " DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu "
"recvd SEGMENT repair request for pending block.\n"); "recvd SEGMENT repair request for pending block.\n");
continue; break;
} }
} }
freshBlock = false;
numErasures = extra_parity; numErasures = extra_parity;
prevBlockId = nextBlockId; prevBlockId = nextBlockId;
} }
@ -2052,21 +2048,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
{ {
DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu " DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu "
"LockSegments() failure\n", LocalNodeId()); "LockSegments() failure\n", LocalNodeId());
inRange = false;
if (!squelchQueued) if (!squelchQueued)
{ {
ServerQueueSquelch(nextObjectId); ServerQueueSquelch(nextObjectId);
squelchQueued = true; squelchQueued = true;
} }
continue; break;
} }
} }
else else
{ {
inRange = false; break; // ignore late arriving NACK
continue;
} }
} // end (object->IsStream() && (nextSegmentId < ndata)) } // end if (object->IsStream() && (nextSegmentId < ndata))
// With a series of SEGMENT repair requests for a block, "numErasures" will // With a series of SEGMENT repair requests for a block, "numErasures" will
// eventually total the number of missing segments in the block. // eventually total the number of missing segments in the block.
@ -2115,12 +2109,11 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
nparity, numErasures); nparity, numErasures);
startTimer = true; startTimer = true;
} // end if/else (holdoff) } // end if/else (holdoff)
inRange = false;
break; break;
case INFO: case INFO:
nextObjectId++; nextObjectId++;
if (nextObjectId > lastObjectId) inRange = false; if (nextObjectId > lastObjectId)
inRange = false;
break; break;
} // end switch(requestLevel) } // end switch(requestLevel)
} // end while(inRange) } // end while(inRange)
@ -2841,7 +2834,7 @@ void NormSession::AdjustRate(bool onResponse)
} }
} }
DMSG(6, "ServerRateTracking time>%lf rate>%lf rtt>%lf loss>%lf\n\n", theTime, tx_rate*(8.0/1000.0), ccRtt, ccLoss); DMSG(8, "ServerRateTracking time>%lf rate>%lf rtt>%lf loss>%lf\n\n", theTime, tx_rate*(8.0/1000.0), ccRtt, ccLoss);
} // end NormSession::AdjustRate() } // end NormSession::AdjustRate()
bool NormSession::OnReportTimeout(ProtoTimer& /*theTimer*/) bool NormSession::OnReportTimeout(ProtoTimer& /*theTimer*/)
@ -2855,8 +2848,16 @@ bool NormSession::OnReportTimeout(ProtoTimer& /*theTimer*/)
if (IsServer()) if (IsServer())
{ {
DMSG(2, "Local status:\n"); DMSG(2, "Local status:\n");
DMSG(2, " txRate>%9.3lf kbps grtt>%lf\n", DMSG(2, " txRate>%9.3lf kbps sentRate>%9.3lf grtt>%lf\n",
((double)tx_rate)*8.0/1000.0, grtt_advertised); ((double)tx_rate)*8.0/1000.0, sent_rate*8.0/1000.0, grtt_advertised);
if (cc_enable)
{
const NormCCNode* clr = (const NormCCNode*)cc_node_list.Head();
if (clr)
DMSG(2, " clr>%lu rate>%9.3lf rtt>%lf loss>%lf\n", clr->GetId(),
clr->GetRate()*8.0/1000.0, clr->GetRtt(), clr->GetLoss());
}
} }
if (IsClient()) if (IsClient())
{ {

3
common/normSession.h
View File

@ -118,6 +118,9 @@ class NormSession
NormSessionMgr& GetSessionMgr() {return session_mgr;} NormSessionMgr& GetSessionMgr() {return session_mgr;}
bool SetRxSocketBuffer(unsigned int bufferSize)
{return rx_socket.SetRxBufferSize(bufferSize);}
// Session parameters // Session parameters
double TxRate() {return (tx_rate * 8.0);} double TxRate() {return (tx_rate * 8.0);}
// (TBD) watch timer scheduling and min/max bounds // (TBD) watch timer scheduling and min/max bounds

128
common/normTest.cpp
View File

@ -16,7 +16,7 @@ int main(int argc, char* argv[])
{ {
printf("normTest starting ...\n"); printf("normTest starting ...\n");
SetDebugLevel(4); SetDebugLevel(2);
NormInstanceHandle instance = NormCreateInstance(); NormInstanceHandle instance = NormCreateInstance();
@ -33,13 +33,13 @@ int main(int argc, char* argv[])
6003, 6003,
NORM_NODE_ANY); NORM_NODE_ANY);
NormSetMessageTrace(session, true); //NormSetMessageTrace(session, true);
NormSetTxLoss(session, 10.0); // 1% packet loss NormSetTxLoss(session, 10.0); // 10% packet loss
NormSetGrttEstimate(session, 0.1);//0.001); // 1 msec initial grtt NormSetGrttEstimate(session, 0.2);//0.001); // 1 msec initial grtt
NormSetTransmitRate(session, 1.0e+05); // in bits/second NormSetTransmitRate(session, 1.0e+06); // in bits/second
NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK); NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK);
@ -60,9 +60,20 @@ int main(int argc, char* argv[])
// Uncomment this line to send a stream instead of the file // Uncomment this line to send a stream instead of the file
stream = NormOpenStream(session, 1024*1024); stream = NormOpenStream(session, 1024*1024);
NormSetStreamFlushMode(stream, NORM_FLUSH_NONE); NormSetStreamFlushMode(stream, NORM_FLUSH_PASSIVE);
int index = -1; // used to monitor reliable stream reception
// Some variable for stream input/output
char txBuffer[8192], rxBuffer[8192];
int txIndex = 0;
int txLen = 0;
int rxIndex = 0;
char refBuffer[1037];
memset(refBuffer, 'a', 1037);
bool msgSync = false;
int msgCount = 0;
int recvCount = -1; // used to monitor reliable stream reception
int sendCount = 0; int sendCount = 0;
int sendMax = 200; int sendMax = 200;
NormEvent theEvent; NormEvent theEvent;
@ -73,17 +84,19 @@ int main(int argc, char* argv[])
case NORM_TX_QUEUE_EMPTY: case NORM_TX_QUEUE_EMPTY:
if (NORM_OBJECT_INVALID != stream) if (NORM_OBJECT_INVALID != stream)
{ {
// Write a message to the "stream" if (0 == txLen)
char buffer[1024];
sprintf(buffer, "normTest says hello %d ...\n", sendCount++);
unsigned int len = strlen(buffer);
TRACE("writing to stream ...\n");
if (len != NormWriteStream(stream, buffer, len))
TRACE("incomplete write:%u\n", len);
else
{ {
NormFlushStream(stream); // Write a message to the "txBuffer"
NormSetWatermark(session, stream); memset(txBuffer, 'a', 1037);
sprintf(txBuffer+1037, "normTest says hello %d ...\n", sendCount);
txLen = strlen(txBuffer);
}
txIndex += NormWriteStream(stream, txBuffer+txIndex, (txLen - txIndex));
if (txIndex == txLen)
{
NormMarkStreamEom(stream);
txLen = txIndex = 0;
sendCount++;
} }
} }
else else
@ -102,9 +115,9 @@ int main(int argc, char* argv[])
else else
{ {
TRACE("QUEUED FILE ...\n"); TRACE("QUEUED FILE ...\n");
sendCount++;
NormSetWatermark(session, txFile); NormSetWatermark(session, txFile);
} }
sendCount++;
} }
break; break;
@ -152,44 +165,77 @@ int main(int argc, char* argv[])
//TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n"); //TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n");
if (NORM_OBJECT_STREAM != NormGetObjectType(theEvent.object)) if (NORM_OBJECT_STREAM != NormGetObjectType(theEvent.object))
break; break;
char buffer[1024]; unsigned int len;
unsigned int len = 1023;
do do
{ {
len = 1023; if (!msgSync)
if (NormReadStream(theEvent.object, buffer, &len)) msgSync = NormFindStreamMsgStart(theEvent.object);
if (!msgSync) break;
len = 8191 - rxIndex;
if (NormReadStream(theEvent.object, rxBuffer+rxIndex, &len))
{ {
buffer[len] = '\0'; rxIndex += len;
if (len) rxBuffer[rxIndex] = '\0';
if (rxIndex > 0)
{ {
TRACE("normTest: recvd(%u):\n\"%s\"\n", len, buffer); char* ptr = strchr(rxBuffer, '\n');
// This while() loop is cheesy test, looking for broken stream if (ptr)
// {
char* ptr = buffer; // Save sub-string length
while ((ptr = strstr(ptr, "hello"))) len = ptr - rxBuffer + 1;
{
int value; // Validate string
if (1 == sscanf(ptr, "hello %d", &value)) if (memcmp(rxBuffer, refBuffer, 1037))
ptr = (char*)NULL;
else
ptr = strstr(rxBuffer, "hello");
if (NULL != ptr)
{ {
if (index >= 0) int value;
if (1 == sscanf(ptr, "hello %d", &value))
{ {
if (1 != (value - index)) if (recvCount >= 0)
TRACE("WARNING! possible break? value:%d index:%d\n", {
value, index); if (1 != (value - recvCount))
TRACE("WARNING! possible break? value:%d recvCount:%d\n",
value, recvCount);
else
msgCount++; // successful recv
//else
// TRACE("validated recv msg len:%d\n", len);
}
recvCount = value;
}
else
{
TRACE("couldn't find index!?\n");
ASSERT(0);
}
if ((unsigned int)rxIndex > len)
{
memmove(rxBuffer, rxBuffer+len, rxIndex - len);
rxIndex -= len;
}
else
{
rxIndex = 0;
} }
index = value;
} }
else else
{ {
TRACE("couldn't find index\n"); TRACE("invalid string!?\n");
ASSERT(0);
} }
ptr += 5;
} }
} }
} }
else else
{ {
TRACE("normTest: error reading stream\n"); TRACE("normTest: error reading stream\n");
TRACE("status: msgCount:%d of total:%d (%lf)\n",
msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount));
msgSync = false;
rxIndex = 0;
break; break;
} }
} while (0 != len); } while (0 != len);
@ -209,7 +255,7 @@ int main(int argc, char* argv[])
} // end switch(theEvent.type) } // end switch(theEvent.type)
// Break after sending "sendMax" messages or files // Break after sending "sendMax" messages or files
if (sendCount >= sendMax) break; //if (sendCount >= sendMax) break;
} // end while (NormGetNextEvent()) } // end while (NormGetNextEvent())

2
common/normVersion.h
View File

@ -32,6 +32,6 @@
#ifndef _NORM_VERSION #ifndef _NORM_VERSION
#define _NORM_VERSION #define _NORM_VERSION
#define VERSION "1.2b2" #define VERSION "1.2b3"
#endif // _NORM_VERSION #endif // _NORM_VERSION