diff --git a/common/normApp.cpp b/common/normApp.cpp index 06bb3fd..71166ca 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -61,6 +61,7 @@ class NormApp : public NormController, public ProtoApp unsigned int input_length; bool input_active; bool push_stream; + NormStreamObject::FlushType msg_flush_mode; bool input_messaging; // stream input mode UINT16 input_msg_length; UINT16 input_msg_index; @@ -110,7 +111,8 @@ NormApp::NormApp() : session_mgr(GetTimerMgr(), GetSocketNotifier()), session(NULL), tx_stream(NULL), rx_stream(NULL), input(NULL), output(NULL), input_index(0), input_length(0), input_active(false), - push_stream(false), input_messaging(false), input_msg_length(0), input_msg_index(0), + push_stream(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE), + 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), address(NULL), port(0), ttl(3), tx_rate(64000.0), cc_enable(false), segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0), @@ -152,6 +154,7 @@ const char* const NormApp::cmd_list[] = "+cc", // congestion control on/off "+rate", // tx date rate (bps) "-push", // push stream writes for real-time messaging + "+flush", // message flushing mode ("none", "passive", or "active") "+input", // send stream input "+output", // recv stream output "+minput", // send message stream input @@ -516,6 +519,21 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val) { push_stream = true; } + else if (!strncmp("flush", cmd, len)) + { + int valLen = strlen(val); + if (!strncmp("none", val, valLen)) + msg_flush_mode = NormStreamObject::FLUSH_NONE; + else if (!strncmp("passive", val, valLen)) + msg_flush_mode = NormStreamObject::FLUSH_PASSIVE; + else if (!strncmp("active", val, valLen)) + msg_flush_mode = NormStreamObject::FLUSH_ACTIVE; + else + { + DMSG(0, "NormApp::ProcessCommand(flush) invalid msg flush mode!\n"); + return false; + } + } else if (!strncmp("processor", cmd, len)) { if (!post_processor->SetCommand(val)) @@ -605,8 +623,8 @@ void NormApp::DoInputReady(ProtoDispatcher::Descriptor /*descriptor*/, void NormApp::OnInputReady() { - //TRACE("NormApp::OnInputReady() ...\n"); - bool flush = false; + //DMSG(0, "NormApp::OnInputReady() ...\n"); + NormStreamObject::FlushType flushType = NormStreamObject::FLUSH_NONE; // write to the stream while input is available _and_ // the stream has buffer space for input while (input) @@ -682,7 +700,7 @@ void NormApp::OnInputReady() } if (stdin != input) fclose(input); input = NULL; - flush = true; + flushType = NormStreamObject::FLUSH_ACTIVE; } else if (ferror(input)) { @@ -705,10 +723,10 @@ void NormApp::OnInputReady() unsigned int writeLength = input_length;// ? input_length - input_index : 0; - if (writeLength || flush) + if (writeLength || (NormStreamObject::FLUSH_NONE != flushType)) { unsigned int wroteLength = tx_stream->Write(input_buffer+input_index, - writeLength, flush, false, + writeLength, flushType, false, push_stream); input_length -= wroteLength; if (0 == input_length) @@ -722,8 +740,8 @@ void NormApp::OnInputReady() { input_msg_index = 0; input_msg_length = 0; - // Mark EOM - tx_stream->Write(NULL, 0, false, true, false); + // Mark EOM _and_ flush + tx_stream->Write(NULL, 0, msg_flush_mode, true, false); } } if (wroteLength < writeLength) @@ -792,7 +810,7 @@ void NormApp::Notify(NormController::Event event, case RX_OBJECT_NEW: { - //TRACE("NormApp::Notify(RX_OBJECT_NEW) ...\n"); + //DMSG(0, "NormApp::Notify(RX_OBJECT_NEW) ...\n"); // It's up to the app to "accept" the object switch (object->GetType()) { @@ -863,7 +881,7 @@ void NormApp::Notify(NormController::Event event, } case RX_OBJECT_INFO: - //TRACE("NormApp::Notify(RX_OBJECT_INFO) ...\n"); + //DMSG(0, "NormApp::Notify(RX_OBJECT_INFO) ...\n"); switch(object->GetType()) { case NormObject::FILE: @@ -903,7 +921,7 @@ void NormApp::Notify(NormController::Event event, break; case RX_OBJECT_UPDATE: - //TRACE("NormApp::Notify(RX_OBJECT_UPDATE) ...\n"); + //DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATE) ...\n"); switch (object->GetType()) { case NormObject::FILE: @@ -949,7 +967,6 @@ void NormApp::Notify(NormController::Event event, findMsgSync = false; } - if(!((NormStreamObject*)object)->Read(output_buffer+output_index, &readLength, findMsgSync)) { @@ -965,11 +982,7 @@ void NormApp::Notify(NormController::Event event, } else { - if (readLength > 0) - output_msg_sync = true; - - TRACE("read %d bytes (index:%d) %hu...\n", readLength, output_index, - *((UINT16*)(output_buffer+output_index))); + if (readLength > 0) output_msg_sync = true; } if (readLength) @@ -1037,7 +1050,7 @@ void NormApp::Notify(NormController::Event event, case RX_OBJECT_COMPLETE: { - //TRACE("NormApp::Notify(RX_OBJECT_COMPLETE) ...\n"); + //DMSG(0, "NormApp::Notify(RX_OBJECT_COMPLETE) ...\n"); switch(object->GetType()) { case NormObject::FILE: @@ -1222,7 +1235,7 @@ bool NormApp::OnStartup(int argc, const char*const* argv) void NormApp::OnShutdown() { - //TRACE("NormApp::OnShutdown() ...\n"); + //DMSG(0, "NormApp::OnShutdown() ...\n"); session_mgr.Destroy(); if (input) { diff --git a/common/normMessage.h b/common/normMessage.h index e99314c..2589ea8 100644 --- a/common/normMessage.h +++ b/common/normMessage.h @@ -369,7 +369,7 @@ class NormMsg UINT16 length; UINT16 header_length; UINT16 header_length_base; - ProtoAddress addr; // src or dst address + ProtoAddress addr; // src or dst address NormMsg* prev; NormMsg* next; diff --git a/common/normObject.cpp b/common/normObject.cpp index 007291a..615b414 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -369,7 +369,7 @@ bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd) { NormBlockId currentId = nextId; nextId++; - //TRACE("NormObject::AppendRepairAdv() testing block:%lu nextId:%lu endId:%lu\n", + //DMSG(0, "NormObject::AppendRepairAdv() testing block:%lu nextId:%lu endId:%lu\n", // (UINT32)currentId, (UINT32)nextId, (UINT32)endId); bool repairEntireBlock = repair_mask.Test(currentId); if (repairEntireBlock) @@ -848,10 +848,10 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, } // end if (appendRequest) nextId++; iterating = GetNextPending(nextId); - //TRACE("got next pending>%lu result:%d\n", (UINT32)nextId, iterating); + //DMSG(0, "got next pending>%lu result:%d\n", (UINT32)nextId, iterating); iterating = iterating && (flush || (nextId <= max_pending_block)); } // end while (iterating || (0 != consecutiveCount)) - //TRACE("bailed ...\n"); + //DMSG(0, "bailed ...\n"); } else { @@ -984,7 +984,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, if (payloadLength < payloadMax) memset(segment+payloadLength, 0, payloadMax-payloadLength); if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) + { segment[payloadMax] = NormObjectMsg::FLAG_MSG_START; + //DMSG(0, "starting segment payload len>%hu msg_len>%hu\n", + // payloadLength, ntohs(*((UINT16*)(segment+8)))) ; + } else segment[payloadMax] = 0; block->AttachSegment(segmentId, segment); @@ -1285,7 +1289,16 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) #ifdef SIMULATE payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); #endif // SIMULATE - if (buffer[payloadMax]) data->SetFlag(NormObjectMsg::FLAG_MSG_START); + if (buffer[payloadMax]) + { + data->SetFlag(NormObjectMsg::FLAG_MSG_START); + //DMSG(0, "read start segment len:%hu\n", ntohs(*((UINT16*)(buffer+2)))); + //DMSG(0, " message len:%hu\n", ntohs(*((UINT16*)(buffer+8)))); + } + else + { + //DMSG(0, "non-start segment read ...\n"); + } } // Perform incremental FEC encoding as needed @@ -1826,7 +1839,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) else { // Stream broken - //TRACE("NormObject::StreamUpdateStatus() broken 1 ...\n"); + //DMSG(0, "NormObject::StreamUpdateStatus() broken 1 ...\n"); return false; } } @@ -1836,7 +1849,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) if (delta > NormBlockId(pending_mask.Size())) { // Stream broken - //TRACE("NormObject::StreamUpdateStatus() broken 2 ...\n"); + //DMSG(0, "NormObject::StreamUpdateStatus() broken 2 ...\n"); return false; } else @@ -1950,7 +1963,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, if (tempOffset == read_offset) { // App didn't want any data here, purge segment - TRACE("NormStreamObject::WriteSegment() app didn't want data ?!\n"); + DMSG(0, "NormStreamObject::WriteSegment() app didn't want data ?!\n"); ASSERT(0); char* s = block->DetachSegment(read_index.segment); segment_pool.Put(s); @@ -1994,7 +2007,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, { char* s = segment_pool.Get(); ASSERT(s != NULL); // for now, this should always succeed - UINT16 payloadLength = NormDataMsg::ReadStreamPayloadLength(segment); + UINT16 payloadLength = NormDataMsg::ReadStreamPayloadLength(segment) + NormDataMsg::GetStreamPayloadHeaderLength(); UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE payloadMax = MIN(SIM_PAYLOAD_MAX, payloadMax); @@ -2002,6 +2015,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, #endif // SIMULATE memcpy(s, segment, payloadLength); s[payloadMax] = segment[payloadMax]; + block->AttachSegment(segmentId, s); block->SetPending(segmentId); ASSERT(block->Segment(segmentId) == s); @@ -2083,6 +2097,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar LocalNodeId(), index, length); read_offset = segmentOffset; *buflen = bytesRead; + ASSERT(0); return false; } @@ -2129,6 +2144,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar #else memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), count); #endif // if/else SIMULATE + index += count; bytesRead += count; read_offset += count; @@ -2153,7 +2169,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar } // end NormStreamObject::Read() unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, - bool flush, bool eom, bool push) + FlushType flushType, bool eom, bool push) { unsigned long nBytes = 0; do @@ -2256,15 +2272,13 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, simCount = MIN(count, simCount); memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, simCount); #else - TRACE("stream write copying %d bytes:index>%d nBytes>%d segment_size>%d space>%d\n", - count, index, nBytes, segment_size, space); memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, count); #endif // if/else SIMULATE NormDataMsg::WriteStreamPayloadLength(segment, index+count); nBytes += count; write_offset += count; // Is the segment full? or flushing - if ((count == space) || ((flush || eom) && (index > 0) && (nBytes == len))) + if ((count == space) || ((FLUSH_NONE != flushType) && (0 != index) && (nBytes == len))) { block->SetPending(write_index.segment); if (++write_index.segment >= ndata) @@ -2280,16 +2294,16 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, { if (eom) msg_start = true; - if (flush) + if (FLUSH_ACTIVE == flushType) flush_pending = true; else flush_pending = false; } else { - flush = false; + flushType = FLUSH_NONE; } - if (nBytes || flush) session->TouchServer(); + if (nBytes || (FLUSH_NONE != flushType)) session->TouchServer(); return nBytes; } // end NormStreamObject::Write() diff --git a/common/normObject.h b/common/normObject.h index 8abc995..1579e0b 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -264,8 +264,18 @@ class NormStreamObject : public NormObject const char* infoPtr = NULL, UINT16 infoLen = 0); void Close(); - bool Accept(unsigned long bufferSize); + + enum FlushType + { + FLUSH_NONE, // no flush action taken + FLUSH_PASSIVE, // pending queued data is transmitted, but no CMD(FLUSH) sent + FLUSH_ACTIVE // pending queued data is transmitted, _and_ active CMD(FLUSH) + }; + bool Read(char* buffer, unsigned int* buflen, bool findMsgStart = false); + unsigned long Write(const char* buffer, unsigned long len, FlushType flushType, bool eom, bool push); + + bool StreamUpdateStatus(NormBlockId blockId); void StreamResync(NormBlockId nextBlockId) {stream_next_id = nextBlockId;} @@ -278,8 +288,6 @@ class NormStreamObject : public NormObject NormSegmentId segmentId, char* buffer); - bool Read(char* buffer, unsigned int* buflen, bool findMsgStart = false); - unsigned long Write(const char* buffer, unsigned long len, bool flush, bool eom, bool push); // For receive stream, we can rewind to earliest buffered offset void Rewind(); diff --git a/common/normSession.cpp b/common/normSession.cpp index 45c2bf2..98fb0eb 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -447,7 +447,7 @@ void NormSession::QueueMessage(NormMsg* msg) double delta = currentTime.tv_sec - lastTime.tv_sec; delta += (((double)currentTime.tv_usec)*1.0e-06 - ((double)lastTime.tv_usec)*1.0e-06); - TRACE("NormSession::QueueMessage() deltaT:%lf\n", delta); + DMSG(0, "NormSession::QueueMessage() deltaT:%lf\n", delta); } lastTime = currentTime; */ @@ -793,7 +793,7 @@ void NormTrace(const struct timeval& currentTime, { const UINT16* x = (const UINT16*)data.GetPayloadData(); if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) - DMSG(0, "start byte>%hu ", ntohs(*x)); + DMSG(0, "start byte>%hu ", ntohs(x[0])); } break; } @@ -1111,7 +1111,7 @@ void NormSession::ServerHandleCCFeedback(NormNodeId nodeId, // adjust rate using current rtt for node ccRate = CalculateRate(nominal_packet_size, ccRtt, ccLoss); } - //TRACE("NormSession::ServerHandleCCFeedback() node>%lu rate>%lf (rtt>%lf loss>%lf slow_start>%d)\n", + //DMSG(0, "NormSession::ServerHandleCCFeedback() node>%lu rate>%lf (rtt>%lf loss>%lf slow_start>%d)\n", // nodeId, ccRate * 8.0 / 1000.0, ccRtt, ccLoss, (0 != (ccFlags & NormCC::START))); // Keep the active CLR (if there is one) at the head of the list @@ -2155,7 +2155,7 @@ bool NormSession::SendMessage(NormMsg& msg) bool drop = (UniformRand(100.0) < tx_loss_rate); if (drop || (clientMsg && client_silent)) { - //TRACE("TX MESSAGE DROPPED! (tx_loss_rate:%lf\n", tx_loss_rate); + //DMSG(0, "TX MESSAGE DROPPED! (tx_loss_rate:%lf\n", tx_loss_rate); } else { @@ -2365,7 +2365,7 @@ void NormSession::AdjustRate(bool onResponse) double scale = tx_rate / sent_rate; scale = MAX(1.0, scale); scale = 1.0; - //TRACE("NormSession::AdjustRate() slow start clr>%lu rate>%lf tx_rate>%lf sent_rate>%lf\n", + //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; } @@ -2383,14 +2383,14 @@ void NormSession::AdjustRate(bool onResponse) tx_rate = clrRate; } } - //TRACE("NormSession::AdjustRate() regular rate adjust clr>%lu rate>%lf (rtt>%lf loss>%lf slow_start>%d)\n", + //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) { // (TBD) fix CC feedback aging ... /*int feedbackAge = abs((int)cc_sequence - (int)clr->GetCCSequence()); - TRACE("NormSession::AdjustRate() feedback age>%d (%d - %d\n", + DMSG(0, "NormSession::AdjustRate() feedback age>%d (%d - %d\n", feedbackAge, cc_sequence, clr->GetCCSequence()); if (feedbackAge > 50) diff --git a/common/normVersion.h b/common/normVersion.h index 34c9b04..eb4617c 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.1b3" +#define VERSION "1.1b4" #endif // _NORM_VERSION