From 541a4bfa9b3a6e22ceca054c7621dd82458a115a Mon Sep 17 00:00:00 2001 From: Jeff Weston Date: Wed, 11 Sep 2019 11:29:44 -0400 Subject: [PATCH] v1.1b2 --- README.md | 6 +- common/normApp.cpp | 7 +- common/normEncoder.cpp | 2 - common/normMessage.h | 80 +++--- common/normNode.cpp | 15 +- common/normObject.cpp | 336 +++++++++++++------------- common/normObject.h | 28 +-- common/normSession.cpp | 67 ++++-- common/normSession.h | 4 +- common/normSimAgent.cpp | 6 +- common/normSimAgent.h | 2 +- common/normVersion.h | 2 +- common/raft.cpp | 521 ++++++++++++++++++++++++++++++++++++++++ unix/Makefile.common | 17 +- win32/Norm.opt | Bin 65024 -> 65024 bytes win32/NormLib.dsp | 40 +-- win32/norm/norm.dsp | 2 +- 17 files changed, 852 insertions(+), 283 deletions(-) create mode 100644 common/raft.cpp diff --git a/README.md b/README.md index f6a2ff6..d3cc679 100644 --- a/README.md +++ b/README.md @@ -6,8 +6,10 @@ command-line application can be built which can send and receive a set of files _or_ a stream piped to/from stdin/stdout. The command- line options are not yet fully documented. -The current code only has Makefiles for Unix platforms, but the -code could be assembled as Win32 project under VC++. +I do have a PDF file which explains some of the command-line options. +Email me for it. + +We're getting there ;-) The NORM code depends upon the current "Protolib" release. See for that code. diff --git a/common/normApp.cpp b/common/normApp.cpp index 5c5bdf2..a1102be 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -123,7 +123,7 @@ NormApp::NormApp() // Init tx_timer for 1.0 second interval, infinite repeats session_mgr.SetController(this); - interval_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormApp::OnIntervalTimeout); + interval_timer.SetListener(this, &NormApp::OnIntervalTimeout); interval_timer.SetInterval(0.0); interval_timer.SetRepeat(0); @@ -707,7 +707,9 @@ void NormApp::OnInputReady() if (writeLength || flush) { - unsigned int wroteLength = tx_stream->Write(input_buffer+input_index, writeLength, flush, false, push_stream); + unsigned int wroteLength = tx_stream->Write(input_buffer+input_index, + writeLength, flush, false, + push_stream); input_length -= wroteLength; if (0 == input_length) input_index = 0; @@ -947,6 +949,7 @@ void NormApp::Notify(NormController::Event event, findMsgSync = false; } + if(!((NormStreamObject*)object)->Read(output_buffer+output_index, &readLength, findMsgSync)) { diff --git a/common/normEncoder.cpp b/common/normEncoder.cpp index 2053698..3b3f5de 100644 --- a/common/normEncoder.cpp +++ b/common/normEncoder.cpp @@ -181,7 +181,6 @@ void NormEncoder::Encode(const char *data, char **pVec) #ifdef SIMULATE UINT16 vecSize = MIN(SIM_PAYLOAD_MAX, vector_size); - vecSize = MAX(vecSize, NormDataMsg::PayloadHeaderLength()); #else UINT16 vecSize = vector_size; #endif // if/else SIMULATE @@ -339,7 +338,6 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era int nvecs = npar + ndata; #ifdef SIMULATE int vecSize = MIN(SIM_PAYLOAD_MAX, vector_size); - vecSize = MAX(NormDataMsg::PayloadHeaderLength(), vecSize); #else int vecSize = vector_size; #endif // if/else SIMUATE diff --git a/common/normMessage.h b/common/normMessage.h index 1846d72..f648fe0 100644 --- a/common/normMessage.h +++ b/common/normMessage.h @@ -10,7 +10,7 @@ #include // for rand(), etc #ifdef SIMULATE -#define SIM_PAYLOAD_MAX 36 // MGEN message size +#define SIM_PAYLOAD_MAX (36+8) // MGEN message size + StreamPayloadHeaderLen() #endif // SIMULATE const UINT8 NORM_PROTOCOL_VERSION = 1; @@ -380,6 +380,7 @@ class NormObjectMsg : public NormMsg UINT8 GetGroupSize() const {return (buffer[GSIZE_OFFSET] & 0x0f);} bool FlagIsSet(NormObjectMsg::Flag flag) const {return (0 != (flag & buffer[FLAGS_OFFSET]));} + bool IsStream() const {return FlagIsSet(FLAG_STREAM);} UINT8 GetFecId() const {return buffer[FEC_ID_OFFSET];} NormObjectId GetObjectId() const { @@ -546,30 +547,28 @@ class NormDataMsg : public NormObjectMsg memcpy(buffer+SYMBOL_ID_OFFSET, &symbolId, 2); } - // Three ways to set payload content: + // Two ways to set payload content: // 1) Directly access payload to copy segment, then set data message length - // (segment must include "payload_len", "payload_offset", and "payload_data" + // (Note NORM_STREAM_OBJECT segments must already include "payload_len" + // and "payload_offset" with the "payload_data" char* AccessPayload() {return (buffer+header_length);} - void SetDataPayloadLength(UINT16 dataLength) + // For NORM_STREAM_OBJECT segments, "dataLength" must include the PAYLOAD_HEADER_LENGTH + void SetPayloadLength(UINT16 payloadLength) { - length = header_length + PAYLOAD_DATA_OFFSET + dataLength; + length = header_length + payloadLength; } - // 2) Set data segment payload with "offset", "data" ptr, and "len" - void SetDataPayload(const NormObjectSize& offset, char* data, UINT16 len) - { - WritePayloadLength(AccessPayload(), len); - WritePayloadOffset(AccessPayload(), offset); - memcpy(buffer+header_length+PAYLOAD_DATA_OFFSET, data, len); - length = header_length + PAYLOAD_DATA_OFFSET + len; - } - // 3) Set "payload" directly (useful for FEC parity segments) + // Set "payload" directly (useful for FEC parity segments) void SetPayload(char* payload, UINT16 payloadLength) { memcpy(buffer+header_length, payload, payloadLength); length = header_length + payloadLength; } - // AccessPayloadData() for ZERO padding - char* AccessPayloadData() {return (buffer+header_length+PAYLOAD_DATA_OFFSET);} + // AccessPayloadData() (useful for setting ZERO padding) + char* AccessPayloadData() + { + UINT16 payloadIndex = IsStream() ? header_length+PAYLOAD_DATA_OFFSET : header_length; + return (buffer+payloadIndex); + } // Message processing methods NormBlockId GetFecBlockId() const @@ -582,36 +581,46 @@ class NormDataMsg : public NormObjectMsg } UINT16 GetFecSymbolId() const - { - return (ntohs(*((UINT16*)(buffer+SYMBOL_ID_OFFSET)))); - } - bool IsData() const {return (GetFecSymbolId() < GetFecBlockLen());} - const char* GetPayloadData() const {return (buffer + header_length + PAYLOAD_DATA_OFFSET);} - UINT16 GetPayloadDataLength() const {return (length - (header_length + PAYLOAD_DATA_OFFSET));} - - // Note: "payload" includes "payload_len", "payload_offset", and "payload_data" fields + {return (ntohs(*((UINT16*)(buffer+SYMBOL_ID_OFFSET))));} + bool IsData() const + {return (GetFecSymbolId() < GetFecBlockLen());} + + // Note: For NORM_OBJECT_STREAM, "payload" includes "payload_reserved", + // "payload_len", "payload_offset", and "payload_data" fields + // For NORM_OBJECT_FILE and NORM_OBJECT_DATA, "payload" includes + // "payload_data" only const char* GetPayload() const {return (buffer + header_length);} UINT16 GetPayloadLength() const {return (length - header_length);} + const char* GetPayloadData() const + { + UINT16 dataIndex = IsStream() ? header_length+PAYLOAD_DATA_OFFSET : header_length; + return (buffer + dataIndex); + } + UINT16 GetPayloadDataLength() const + { + UINT16 dataIndex = IsStream() ? header_length+PAYLOAD_DATA_OFFSET : header_length; + return (length - dataIndex); + } + + // These routines are only applicable to messages containing NORM_OBJECT_STREAM content // Some static helper routines for reading/writing embedded payload length/offsets - static UINT16 PayloadHeaderLength() {return (PAYLOAD_DATA_OFFSET);} - static void WritePayloadLength(char* payload, UINT16 len) + static UINT16 GetStreamPayloadHeaderLength() {return (PAYLOAD_DATA_OFFSET);} + static void WriteStreamPayloadLength(char* payload, UINT16 len) { *((UINT16*)(payload+PAYLOAD_LENGTH_OFFSET)) = htons(len); } - static void WritePayloadOffset(char* payload, const NormObjectSize& offset) + static void WriteStreamPayloadOffset(char* payload, UINT32 offset) { - *((UINT16*)(payload+PAYLOAD_OFFSET_OFFSET)) = htons(offset.MSB()); - *((UINT32*)(payload+PAYLOAD_OFFSET_OFFSET+2)) = htonl(offset.LSB()); + *((UINT32*)(payload+PAYLOAD_OFFSET_OFFSET)) = htonl(offset); } - static UINT16 ReadPayloadLength(const char* payload) + static UINT16 ReadStreamPayloadLength(const char* payload) { return (ntohs(*((UINT16*)(payload+PAYLOAD_LENGTH_OFFSET)))); } - static NormObjectSize ReadPayloadOffset(const char* payload) + static UINT32 ReadStreamPayloadOffset(const char* payload) { - return NormObjectSize(ntohs(*((UINT16*)(payload+PAYLOAD_OFFSET_OFFSET))), - ntohl(*((UINT32*)(payload+PAYLOAD_OFFSET_OFFSET+2)))); + return ntohl(*((UINT32*)(payload+PAYLOAD_OFFSET_OFFSET))); } private: @@ -624,9 +633,10 @@ class NormDataMsg : public NormObjectMsg }; enum { - PAYLOAD_LENGTH_OFFSET = 0, + PAYLOAD_RESERVED_OFFSET = 0, + PAYLOAD_LENGTH_OFFSET = PAYLOAD_RESERVED_OFFSET + 2, PAYLOAD_OFFSET_OFFSET = PAYLOAD_LENGTH_OFFSET + 2, - PAYLOAD_DATA_OFFSET = PAYLOAD_OFFSET_OFFSET + 6 + PAYLOAD_DATA_OFFSET = PAYLOAD_OFFSET_OFFSET + 4 }; }; // end class NormDataMsg diff --git a/common/normNode.cpp b/common/normNode.cpp index 349fee3..79b1546 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -37,15 +37,15 @@ NormServerNode::NormServerNode(class NormSession* theSession, NormNodeId nodeId) recv_total(0), recv_goodput(0), resync_count(0), nack_count(0), suppress_count(0), completion_count(0), failure_count(0) { - repair_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormServerNode::OnRepairTimeout); + repair_timer.SetListener(this, &NormServerNode::OnRepairTimeout); repair_timer.SetInterval(0.0); repair_timer.SetRepeat(1); - activity_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormServerNode::OnActivityTimeout); + activity_timer.SetListener(this, &NormServerNode::OnActivityTimeout); activity_timer.SetInterval(NormSession::DEFAULT_GRTT_ESTIMATE*NORM_ROBUST_FACTOR); activity_timer.SetRepeat(NORM_ROBUST_FACTOR); - cc_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormServerNode::OnCCTimeout); + cc_timer.SetListener(this, &NormServerNode::OnCCTimeout); cc_timer.SetInterval(0.0); cc_timer.SetRepeat(1); @@ -101,7 +101,7 @@ bool NormServerNode::Open(UINT16 sessionId, UINT16 segmentSize, UINT16 numData, unsigned long blockSpace = sizeof(NormBlock) + blockSize * sizeof(char*) + 2*maskSize + - numData * (segmentSize + NormDataMsg::PayloadHeaderLength()); + numData * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength()); unsigned long bufferSpace = session->RemoteServerBufferSize(); unsigned long numBlocks = bufferSpace / blockSpace; if (bufferSpace > (numBlocks*blockSpace)) numBlocks++; @@ -118,14 +118,14 @@ bool NormServerNode::Open(UINT16 sessionId, UINT16 segmentSize, UINT16 numData, // The extra byte of segments is used for marking segments // which are "start segments" for messages encapsulated in // a NormStreamObject - if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::PayloadHeaderLength()+1)) + if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()+1)) { DMSG(0, "NormServerNode::Open() segment_pool init error\n"); Close(); return false; } - if (!decoder.Init(numParity, segmentSize+NormDataMsg::PayloadHeaderLength())) + if (!decoder.Init(numParity, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength())) { DMSG(0, "NormServerNode::Open() decoder init error: %s\n", strerror(errno)); @@ -166,6 +166,7 @@ void NormServerNode::Close() rx_table.Destroy(); segment_size = ndata = nparity = 0; is_open = false; + synchronized = false; } // end NormServerNode::Close() @@ -709,6 +710,8 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) if (!IsOpen()) { + DMSG(4, "NormServerNode::HandleObjectMessage() node>%lu opening server>%lu ...\n", + LocalNodeId(), GetId()); // Currently,, our implementation requires the FEC Object Transmission Information // to properly allocate resources NormFtiExtension fti; diff --git a/common/normObject.cpp b/common/normObject.cpp index b0283ae..912a6d9 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -94,10 +94,6 @@ bool NormObject::Open(const NormObjectSize& objectSize, NormObjectSize numBlocks = numSegments / NormObjectSize(numData); ASSERT(0 == numBlocks.MSB()); - - - - if (!block_buffer.Init(numBlocks.LSB())) { DMSG(0, "NormObject::Open() init block_buffer error\n"); @@ -212,7 +208,8 @@ bool NormObject::HandleBlockRequest(NormBlockId nextId, NormBlockId lastId) DMSG(6, "NormObject::HandleBlockRequest() node>%lu blk>%lu -> blk>%lu\n", LocalNodeId(), (UINT32)nextId, (UINT32)lastId); bool increasedRepair = false; - while (nextId <= lastId) + lastId++; + while (nextId != lastId) { if (!repair_mask.Test(nextId)) { @@ -267,7 +264,8 @@ bool NormObject::TxResetBlocks(NormBlockId nextId, NormBlockId lastId) { bool increasedRepair = false; UINT16 autoParity = session->ServerAutoParity(); - while (nextId <= lastId) + lastId++; + while (nextId != lastId) { if (!pending_mask.Test(nextId)) { @@ -282,8 +280,6 @@ bool NormObject::TxResetBlocks(NormBlockId nextId, NormBlockId lastId) return increasedRepair; } // end NormObject::TxResetBlocks() - - bool NormObject::ActivateRepairs() { bool repairsActivated = false; @@ -949,8 +945,8 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, (UINT16)id); return; } - UINT16 segmentLen = data.GetPayloadDataLength(); - if (segmentLen > segment_size) + UINT16 segmentLength = data.GetPayloadDataLength(); + if (segmentLength > segment_size) { DMSG(0, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " "Error! segment too large ...\n", LocalNodeId(), server->GetId(), @@ -958,28 +954,19 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, server->PutFreeSegment(segment); return; } + UINT16 payloadLength = data.GetPayloadLength(); + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - // For simulations, we just need the payload header (size & offset) - UINT16 simLen = MIN(segmentLen, SIM_PAYLOAD_MAX); - memcpy(segment, data.GetPayload(), simLen+NormDataMsg::PayloadHeaderLength()); - simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - if (segmentLen < simLen) - memset(segment+segmentLen+NormDataMsg::PayloadHeaderLength(), 0, simLen - segmentLen); + // For simulations, we may need to cap the payloadLength + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + if (payloadLength < payloadMax) + memset(segment+payloadLength, 0, payloadMax-payloadLength); + memcpy(segment, data.GetPayload(), payloadLength); if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) - segment[simLen+NormDataMsg::PayloadHeaderLength()] = - NormObjectMsg::FLAG_MSG_START; + segment[payloadMax] = NormObjectMsg::FLAG_MSG_START; else - segment[simLen+NormDataMsg::PayloadHeaderLength()] = 0; -#else - memcpy(segment, data.GetPayload(), segmentLen+NormDataMsg::PayloadHeaderLength()); - if (segmentLen < segment_size) - memset(segment+segmentLen+NormDataMsg::PayloadHeaderLength(), 0, segment_size - segmentLen); - if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) - segment[segment_size+NormDataMsg::PayloadHeaderLength()] = - NormObjectMsg::FLAG_MSG_START; - else - segment[segment_size+NormDataMsg::PayloadHeaderLength()] = 0; -#endif // if/else SIMULATE + segment[payloadMax] = 0; block->AttachSegment(segmentId, segment); block->UnsetPending(segmentId); // 2) Write segment to object (if it's data) @@ -987,7 +974,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, { block->DecrementErasureCount(); if (WriteSegment(blockId, segmentId, segment)) - server->IncrementRecvGoodput(segmentLen); + server->IncrementRecvGoodput(segmentLength); } else { @@ -1019,12 +1006,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, // (TBD) Dump the block ...??? return; } -#ifdef SIMULATE - UINT16 simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - memset(segment, 0, simLen + NormDataMsg::PayloadHeaderLength()+1); -#else - memset(segment, 0, segment_size+NormDataMsg::PayloadHeaderLength()+1); -#endif // !SIMULATE + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); +#ifdef SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + memset(segment, 0, payloadMax+1); block->SetSegment(nextErasure, segment); } nextErasure = block->NextPending(nextErasure+1); @@ -1041,7 +1027,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, if (WriteSegment(blockId, sid, block->Segment(sid))) { // For statistics only (TBD) #ifdef NORM_DEBUG - server->IncrementRecvGoodput(NormDataMsg::ReadPayloadLength(block->Segment(sid))); + server->IncrementRecvGoodput(segmentLength); } } else @@ -1224,12 +1210,11 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) char* s = session->ServerGetFreeSegment(id, blockId); if (s) { + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - UINT16 simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - memset(s, 0, simLen + NormDataMsg::PayloadHeaderLength()+1); -#else - memset(s, 0, segment_size + NormDataMsg::PayloadHeaderLength()+1); -#endif // if/else SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + memset(s, 0, payloadMax+1); // extra byte for msg flags block->AttachSegment(i, s); } else @@ -1260,30 +1245,26 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) // Try to read segment if (segmentId < numData) { - // Try to read data segment + // Try to read data segment (Note "ReadSegment" copies in offset/length info also) char* buffer = data->AccessPayload(); - if (!ReadSegment(blockId, segmentId, buffer)) + UINT16 payloadLength = ReadSegment(blockId, segmentId, buffer); + if (0 == payloadLength) { // (TBD) deal with read error //(for streams, it currently means the stream is non-pending) - if (!IsStream()) - DMSG(0, "NormObject::NextServerMsg() ReadSegment() error\n"); + if (!IsStream()) DMSG(0, "NormObject::NextServerMsg() ReadSegment() error\n"); return false; } - UINT16 length = NormDataMsg::ReadPayloadLength(buffer); - data->SetDataPayloadLength(length); + data->SetPayloadLength(payloadLength); if (IsStream()) { // Look for FLAG_MSG_START + UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - UINT16 simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - if (buffer[simLen+NormDataMsg::PayloadHeaderLength()]) - data->SetFlag(NormObjectMsg::FLAG_MSG_START); -#else - if (buffer[segment_size+NormDataMsg::PayloadHeaderLength()]) - data->SetFlag(NormObjectMsg::FLAG_MSG_START); -#endif // if/else SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + if (buffer[payloadMax]) data->SetFlag(NormObjectMsg::FLAG_MSG_START); } // Perform incremental FEC encoding as needed @@ -1292,8 +1273,12 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) { // (TBD) for non-stream objects, catch alternate "last block/segment len" // ZERO pad any "runt" segments before encoding - if (length < segment_size) - memset(buffer+NormDataMsg::PayloadHeaderLength()+length, 0, segment_size-length); + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); +#ifdef SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + if (payloadLength < payloadMax) + memset(buffer+payloadLength, 0, payloadMax-payloadLength); // (TBD) the encode routine could update the block's parity readiness session->ServerEncode(data->AccessPayload(), block->SegmentList(numData)); block->IncreaseParityReadiness(); @@ -1309,7 +1294,9 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) ASSERT(block->ParityReady(numData)); char* segment = block->Segment(segmentId); ASSERT(segment); - data->SetPayload(segment, segment_size+NormDataMsg::PayloadHeaderLength()); + UINT16 payloadLength = segment_size; + if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength(); + data->SetPayload(segment, payloadLength); } block->UnsetPending(segmentId); if (block->InRepair()) @@ -1348,6 +1335,7 @@ void NormStreamObject::StreamAdvance() ASSERT(block); if (!block->IsTransmitPending()) { + // ??? (TBD) Should this block be returned to the pool right now??? if (pending_mask.Set(nextBlockId)) stream_next_id++; else @@ -1372,11 +1360,15 @@ bool NormObject::CalculateBlockParity(NormBlock* block) UINT16 numData = GetBlockSize(block->Id()); for (UINT16 i = 0; i < numData; i++) { - if (ReadSegment(block->Id(), i, buffer)) + UINT16 payloadLength = ReadSegment(block->Id(), i, buffer); + if (0 != payloadLength) { - UINT16 length = NormDataMsg::ReadPayloadLength(buffer); - if (length < segment_size) - memset(buffer+NormDataMsg::PayloadHeaderLength()+length, 0, segment_size-length); + UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); +#ifdef SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + if (payloadLength < payloadMax) + memset(buffer+payloadLength, 0, payloadMax-payloadLength); session->ServerEncode(buffer, block->SegmentList(numData)); } else @@ -1403,12 +1395,11 @@ NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId) char* s = session->ServerGetFreeSegment(id, blockId); if (s) { + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - UINT16 simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - memset(s, 0, simLen + NormDataMsg::PayloadHeaderLength()); -#else - memset(s, 0, segment_size + NormDataMsg::PayloadHeaderLength()); -#endif // if/else SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + memset(s, 0, payloadMax+1); // extra byte for msg flags block->AttachSegment(i, s); } else @@ -1511,8 +1502,6 @@ bool NormFileObject::Open(const char* thePath, file.Close(); return false; } - large_block_length = NormObjectSize(large_block_size) * segment_size; - small_block_length = NormObjectSize(small_block_size) * segment_size; } else { @@ -1520,6 +1509,8 @@ bool NormFileObject::Open(const char* thePath, return false; } } + large_block_length = NormObjectSize(large_block_size) * segment_size; + small_block_length = NormObjectSize(small_block_size) * segment_size; strncpy(path, thePath, PATH_MAX); unsigned int len = strlen(thePath); len = MIN(len, PATH_MAX); @@ -1562,23 +1553,36 @@ bool NormFileObject::WriteSegment(NormBlockId blockId, { len = segment_size; } - NormObjectSize segmentOffset = NormDataMsg::ReadPayloadOffset(buffer); + // Determine segment offset from blockId::segmentId + NormObjectSize segmentOffset; + NormObjectSize segmentSize = NormObjectSize(segment_size); + if ((UINT32)blockId < large_block_count) + { + segmentOffset = large_block_length*(UINT32)blockId + segmentSize*segmentId; + } + else + { + segmentOffset = large_block_length*large_block_count; // (TBD) pre-calc this + UINT32 smallBlockIndex = (UINT32)blockId - large_block_count; + segmentOffset = segmentOffset + small_block_length*smallBlockIndex + + segmentSize*segmentId; + } off_t offsetScaleMSB = 0xffffffff + 1; off_t offset = (off_t)segmentOffset.LSB() + ((off_t)segmentOffset.MSB() * offsetScaleMSB); if (offset != file.GetOffset()) { if (!file.Seek(offset)) return false; } - UINT16 nbytes = file.Write(buffer+NormDataMsg::PayloadHeaderLength(), len); + UINT16 nbytes = file.Write(buffer, len); return (nbytes == len); } // end NormFileObject::WriteSegment() -bool NormFileObject::ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer) +UINT16 NormFileObject::ReadSegment(NormBlockId blockId, + NormSegmentId segmentId, + char* buffer) { - // Determine segment length + // Determine segment length from blockId::segmentId UINT16 len; if (blockId == final_block_id) { @@ -1591,6 +1595,8 @@ bool NormFileObject::ReadSegment(NormBlockId blockId, { len = segment_size; } + + // Determine segment offset from blockId::segmentId NormObjectSize segmentOffset; NormObjectSize segmentSize = NormObjectSize(segment_size); @@ -1600,13 +1606,11 @@ bool NormFileObject::ReadSegment(NormBlockId blockId, } else { - segmentOffset = large_block_length*large_block_count; + segmentOffset = large_block_length*large_block_count; // (TBD) pre-calc this UINT32 smallBlockIndex = (UINT32)blockId - large_block_count; segmentOffset = segmentOffset + small_block_length*smallBlockIndex + - segmentSize *segmentId; + segmentSize*segmentId; } - NormDataMsg::WritePayloadLength(buffer, len); - NormDataMsg::WritePayloadOffset(buffer, segmentOffset); off_t offsetScaleMSB = 0xffffffff + 1; off_t offset = (off_t)segmentOffset.LSB() + ((off_t)segmentOffset.MSB() * offsetScaleMSB); if (offset != file.GetOffset()) @@ -1614,8 +1618,8 @@ bool NormFileObject::ReadSegment(NormBlockId blockId, if (!file.Seek(offset)) return false; } - UINT16 nbytes = file.Read(buffer+NormDataMsg::PayloadHeaderLength(), len); - return (len == nbytes); + UINT16 nbytes = file.Read(buffer, len); + return (len == nbytes) ? len : 0; } // end NormFileObject::ReadSegment() @@ -1673,7 +1677,7 @@ bool NormStreamObject::Open(unsigned long bufferSize, return false; } - if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::PayloadHeaderLength()+1)) + if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()+1)) { DMSG(0, "NormStreamObject::Open() segment_pool init error\n"); Close(); @@ -1690,7 +1694,7 @@ bool NormStreamObject::Open(unsigned long bufferSize, // since our objects are exclusively read _or_ write write_index.block = write_index.segment = 0; read_index.block = read_index.segment = 0; - write_offset = read_offset = NormObjectSize((UINT32)0); + write_offset = read_offset = 0; if (!server) { if (!NormObject::Open(NormObjectSize((UINT32)bufferSize), infoPtr, infoLen)) @@ -1721,7 +1725,7 @@ bool NormStreamObject::Accept(unsigned long bufferSize) void NormStreamObject::Close() { NormObject::Close(); - write_offset = read_offset = NormObjectSize((UINT32)0); + write_offset = read_offset = 0; NormBlock* b; while ((b = stream_buffer.Find(stream_buffer.RangeLo()))) { @@ -1844,9 +1848,9 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) } // end NormStreamObject::StreamUpdateStatus() -bool NormStreamObject::ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer) +UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, + NormSegmentId segmentId, + char* buffer) { // (TBD) compare blockId with stream_buffer.RangeLo() and stream_buffer.RangeHi() NormBlock* block = stream_buffer.Find(blockId); @@ -1864,19 +1868,19 @@ bool NormStreamObject::ReadSegment(NormBlockId blockId, block->UnsetPending(segmentId); char* segment = block->Segment(segmentId); ASSERT(segment != NULL); - UINT16 length = NormDataMsg::ReadPayloadLength(segment); - ASSERT(length <= segment_size); + UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); + ASSERT(segmentLength <= segment_size); #ifdef SIMULATE - length = MIN(length, SIM_PAYLOAD_MAX); - UINT16 simLen = MIN(SIM_PAYLOAD_MAX, segment_size); - buffer[simLen+NormDataMsg::PayloadHeaderLength()] = - segment[simLen+NormDataMsg::PayloadHeaderLength()]; + UINT16 simLen = segmentLength + NormDataMsg::GetStreamPayloadHeaderLength(); + simLen = MIN(simLen, SIM_PAYLOAD_MAX); + buffer[simLen] = segment[simLen]; #else - buffer[segment_size+NormDataMsg::PayloadHeaderLength()] = - segment[segment_size+NormDataMsg::PayloadHeaderLength()]; + buffer[segment_size+NormDataMsg::GetStreamPayloadHeaderLength()] = + segment[segment_size+NormDataMsg::GetStreamPayloadHeaderLength()]; #endif // if/else SIMULATE - memcpy(buffer, segment, length+NormDataMsg::PayloadHeaderLength()); - return true; + UINT16 payloadLength = segmentLength+NormDataMsg::GetStreamPayloadHeaderLength(); + memcpy(buffer, segment, payloadLength); + return payloadLength; } // end NormStreamObject::ReadSegment() bool NormStreamObject::WriteSegment(NormBlockId blockId, @@ -1889,13 +1893,15 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, { //DMSG(0, "NormStreamObject::WriteSegment() block/segment < read_index!?\n"); return false; - }*/ + }*/ - - NormObjectSize segmentOffset = NormDataMsg::ReadPayloadOffset(segment); - if (read_offset > segmentOffset) + UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); + // if (segmentOffset < read_offset) + UINT32 diff = segmentOffset - read_offset; + if ((diff > 0x80000000) || ((0x80000000 == diff) && (segmentOffset > read_offset))) { - //DMSG(0, "NormStreamObject::WriteSegment() segmentOffset < read_offset!?\n"); + DMSG(0, "NormStreamObject::WriteSegment() diff:%lu segmentOffset:%lu < read_offset:%lu \n", + diff, segmentOffset, read_offset); return false; } @@ -1916,7 +1922,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, read_index.block = block->Id(); read_index.segment = block->FirstPending(); NormBlock* tempBlock = block; - NormObjectSize tempOffset = read_offset; + UINT32 tempOffset = read_offset; session->Notify(NormController::RX_OBJECT_UPDATE, server, this); block = stream_buffer.Find(stream_buffer.RangeLo()); if (tempBlock == block) @@ -1968,17 +1974,14 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, { char* s = segment_pool.Get(); ASSERT(s != NULL); // for now, this should always succeed - UINT16 length = NormDataMsg::ReadPayloadLength(segment); + UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); + UINT16 flagsOffset = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - length = MIN(length, SIM_PAYLOAD_MAX); - UINT16 simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - s[simLen+NormDataMsg::PayloadHeaderLength()] = - segment[simLen+NormDataMsg::PayloadHeaderLength()]; -#else - s[segment_size+NormDataMsg::PayloadHeaderLength()] = - segment[segment_size+NormDataMsg::PayloadHeaderLength()]; + flagsOffset = MIN(SIM_PAYLOAD_MAX, flagsOffset); + segmentLength = MIN((SIM_PAYLOAD_MAX-NormDataMsg::GetStreamPayloadHeaderLength()), segmentLength); #endif // SIMULATE - memcpy(s, segment, length + NormDataMsg::PayloadHeaderLength()); + s[flagsOffset] = segment[flagsOffset]; + memcpy(s, segment, segmentLength + NormDataMsg::GetStreamPayloadHeaderLength()); block->AttachSegment(segmentId, s); block->SetPending(segmentId); ASSERT(block->Segment(segmentId) == s); @@ -2035,26 +2038,25 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar char* segment = block->Segment(read_index.segment); if (!segment) { - //DMSG(0, "NormStreamObject::Read() stream buffer empty (2)\n"); + //DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n", + // (UINT32)read_index.block, read_index.segment); *buflen = bytesRead; return true; } - NormObjectSize segmentOffset = NormDataMsg::ReadPayloadOffset(segment); - if (segmentOffset > read_offset) + UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); + // if (read_offset < segmentOffset) + UINT32 diff = read_offset - segmentOffset; + if ((diff > 0x80000000) || ((0x80000000 == diff) && (read_offset > segmentOffset))) { - DMSG(4, "NormStreamObject::Read() node>%lu broken stream!\n", LocalNodeId()); + DMSG(0, "NormStreamObject::Read() node>%lu broken stream!\n", LocalNodeId()); read_offset = segmentOffset; *buflen = bytesRead; return false; } - NormObjectSize delta = read_offset - segmentOffset; - - ASSERT(!delta.MSB()); - UINT16 index = delta.LSB(); - UINT16 length = NormDataMsg::ReadPayloadLength(segment); - + UINT32 index = read_offset - segmentOffset; + UINT16 length = NormDataMsg::ReadStreamPayloadLength(segment); if (index >= length) { DMSG(0, "NormStreamObject::Read() node>%lu mangled stream! index:%hu length:%hu\n", @@ -2077,14 +2079,11 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar } else { + UINT16 flagsOffset = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - UINT16 simLen = MIN(segment_size, SIM_PAYLOAD_MAX); - msgStart = (NormObjectMsg::FLAG_MSG_START == - segment[simLen+NormDataMsg::PayloadHeaderLength()]); -#else - msgStart = (NormObjectMsg::FLAG_MSG_START == - segment[segment_size+NormDataMsg::PayloadHeaderLength()]); + flagsOffset = MIN(flagsOffset, SIM_PAYLOAD_MAX); #endif // if/else SIMULATE + msgStart = (NormObjectMsg::FLAG_MSG_START == segment[flagsOffset]); } if (!msgStart) { @@ -2107,9 +2106,9 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar UINT16 simCount = ((index+count) < SIM_PAYLOAD_MAX) ? count : ((index < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - index) : 0); - memcpy(buffer+bytesRead, segment+index+NormDataMsg::PayloadHeaderLength(), simCount); + memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), simCount); #else - memcpy(buffer+bytesRead, segment+index+NormDataMsg::PayloadHeaderLength(), count); + memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), count); #endif // if/else SIMULATE index += count; bytesRead += count; @@ -2134,10 +2133,11 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar return true; } // end NormStreamObject::Read() -unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, bool flush, bool eom, bool push) +unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, + bool flush, bool eom, bool push) { unsigned long nBytes = 0; - while (nBytes < len) + do { NormBlock* block = stream_buffer.Find(write_index.block); if (!block) @@ -2201,33 +2201,32 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, boo segment = segment_pool.Get(); ASSERT(segment); } - NormDataMsg::WritePayloadOffset(segment, write_offset); - NormDataMsg::WritePayloadLength(segment, 0); + NormDataMsg::WriteStreamPayloadOffset(segment, write_offset); + NormDataMsg::WriteStreamPayloadLength(segment, 0); block->AttachSegment(write_index.segment, segment); } - UINT16 index = NormDataMsg::ReadPayloadLength(segment); + UINT16 index = NormDataMsg::ReadStreamPayloadLength(segment); if (0 == index) { - // On first write to segment, mark MSG_START flag as appropriate - char msgStartValue; - if (msg_start) + char msgStartValue = 0; + if (msg_start && len) { + ASSERT(0 == index); + msgStartValue = NormObjectMsg::FLAG_MSG_START; msg_start = false; - msgStartValue = NormObjectMsg::FLAG_MSG_START; - } - else - { - msgStartValue = 0; } + UINT16 flagsOffset = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - UINT16 simLen = MIN(SIM_PAYLOAD_MAX, segment_size); - segment[simLen+NormDataMsg::PayloadHeaderLength()] = msgStartValue; -#else - segment[segment_size+NormDataMsg::PayloadHeaderLength()] = msgStartValue; -#endif // if/else SIMULATE + flagsOffset = MIN(SIM_PAYLOAD_MAX, flagsOffset); +#endif // SIMULATE + segment[flagsOffset] = msgStartValue; } + else + { + msg_start = false; + } UINT16 count = (UINT16)(len - nBytes); UINT16 space = segment_size - index; @@ -2236,17 +2235,16 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, boo UINT16 simCount = ((index+count) < SIM_PAYLOAD_MAX) ? count : ((index < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - index) : 0); - memcpy(segment+index+NormDataMsg::PayloadHeaderLength(), buffer+nBytes, simCount); + memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, simCount); #else - memcpy(segment+index+NormDataMsg::PayloadHeaderLength(), buffer+nBytes, count); + memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, count); #endif // if/else SIMULATE - NormDataMsg::WritePayloadLength(segment, index+count); + NormDataMsg::WriteStreamPayloadLength(segment, index+count); nBytes += count; write_offset += count; - // Is the segment full? or flushing, or end-of-message - if ((count == space) || ((flush || eom) && (index > 0))) + // Is the segment full? or flushing + if ((count == space) || ((flush || eom) && (index > 0) && (nBytes == len))) { - if (eom) ASSERT(0); block->SetPending(write_index.segment); if (++write_index.segment >= ndata) { @@ -2254,13 +2252,13 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, boo write_index.segment = 0; } } - } - - // if this was end-of-message next Write() will be considered a new message + } while (nBytes < len); + // if this was end-of-message next Write() will be considered a new message if (nBytes == len) { - if (eom) msg_start = true; + if (eom) + msg_start = true; if (flush) flush_pending = true; else @@ -2293,20 +2291,24 @@ NormSimObject::~NormSimObject() } -bool NormSimObject::ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer) +UINT16 NormSimObject::ReadSegment(NormBlockId blockId, + NormSegmentId segmentId, + char* buffer) { // Determine segment length UINT16 len; - if ((blockId == last_block_id) && - (segmentId == (last_block_size - 1))) - len = last_segment_size; + if (blockId == final_block_id) + { + if (segmentId == (GetBlockSize(blockId)-1)) + len = final_segment_size; + else + len = segment_size; + } else + { len = segment_size; - // the "len" is needed to build the correct size message - NormDataMsg::WritePayloadLength(buffer, len); - return true; + } + return len; } // end NormSimObject::ReadSegment() #endif // SIMULATE diff --git a/common/normObject.h b/common/normObject.h index d1a574f..9006815 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -55,9 +55,9 @@ class NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, const char* buffer) = 0; - virtual bool ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer) = 0; + virtual UINT16 ReadSegment(NormBlockId blockId, + NormSegmentId segmentId, + char* buffer) = 0; // These are only valid after object is open NormBlockId GetFinalBlockId() const {return final_block_id;} @@ -197,9 +197,9 @@ class NormFileObject : public NormObject NormSegmentId segmentId, const char* buffer); - virtual bool ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer); + virtual UINT16 ReadSegment(NormBlockId blockId, + NormSegmentId segmentId, + char* buffer); private: char path[PATH_MAX]; @@ -231,9 +231,9 @@ class NormStreamObject : public NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, const char* buffer); - virtual bool ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer); + virtual UINT16 ReadSegment(NormBlockId blockId, + 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); @@ -271,9 +271,9 @@ class NormStreamObject : public NormObject NormSegmentPool segment_pool; NormBlockBuffer stream_buffer; Index write_index; - NormObjectSize write_offset; + UINT32 write_offset; Index read_index; - NormObjectSize read_offset; + UINT32 read_offset; bool flush_pending; bool msg_start; }; // end class NormStreamObject @@ -302,9 +302,9 @@ class NormSimObject : public NormObject NormSegmentId segmentId, const char* buffer) {return true;} - virtual bool ReadSegment(NormBlockId blockId, - NormSegmentId segmentId, - char* buffer); + virtual UINT16 ReadSegment(NormBlockId blockId, + NormSegmentId segmentId, + char* buffer); }; // end class NormSimObject #endif // SIMULATE diff --git a/common/normSession.cpp b/common/normSession.cpp index 28e0f2b..2cb520f 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -26,7 +26,8 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) flush_count(NORM_ROBUST_FACTOR+1), posted_tx_queue_empty(false), advertise_repairs(false), suppress_nonconfirmed(false), suppress_rate(-1.0), suppress_rtt(-1.0), - probe_proactive(true), grtt_interval(0.5), + probe_proactive(true), probe_pending(false), probe_reset(false), + grtt_interval(0.5), grtt_interval_min(DEFAULT_GRTT_INTERVAL_MIN), grtt_interval_max(DEFAULT_GRTT_INTERVAL_MAX), grtt_max(DEFAULT_GRTT_MAX), @@ -38,24 +39,24 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) next(NULL) { tx_socket.SetNotifier(&sessionMgr.GetSocketNotifier()); - tx_socket.SetListener(this, (ProtoSocket::EventHandler)&NormSession::TxSocketRecvHandler); + tx_socket.SetListener(this, &NormSession::TxSocketRecvHandler); rx_socket.SetNotifier(&sessionMgr.GetSocketNotifier()); - rx_socket.SetListener(this, (ProtoSocket::EventHandler)&NormSession::RxSocketRecvHandler); + rx_socket.SetListener(this, &NormSession::RxSocketRecvHandler); - tx_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormSession::OnTxTimeout); + tx_timer.SetListener(this, &NormSession::OnTxTimeout); tx_timer.SetInterval(0.0); tx_timer.SetRepeat(-1); - repair_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormSession::OnRepairTimeout); + repair_timer.SetListener(this, &NormSession::OnRepairTimeout); repair_timer.SetInterval(0.0); repair_timer.SetRepeat(1); - flush_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormSession::OnFlushTimeout); + flush_timer.SetListener(this, &NormSession::OnFlushTimeout); flush_timer.SetInterval(0.0); flush_timer.SetRepeat(0); - probe_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormSession::OnProbeTimeout); + probe_timer.SetListener(this, &NormSession::OnProbeTimeout); probe_timer.SetInterval(0.0); probe_timer.SetRepeat(-1); @@ -70,7 +71,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) // This timer is for printing out occasional status reports // (It may be used to trigger transmission of report messages // in the future for debugging, etc - report_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormSession::OnReportTimeout); + report_timer.SetListener(this, &NormSession::OnReportTimeout); report_timer.SetInterval(10.0); report_timer.SetRepeat(-1); } @@ -189,7 +190,7 @@ bool NormSession::StartServer(unsigned long bufferSpace, unsigned long blockSpace = sizeof(NormBlock) + blockSize * sizeof(char*) + 2*maskSize + - numParity * (segmentSize + NormDataMsg::PayloadHeaderLength()); + numParity * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength()); unsigned long numBlocks = bufferSpace / blockSpace; if (bufferSpace > (numBlocks*blockSpace)) numBlocks++; @@ -204,7 +205,7 @@ bool NormSession::StartServer(unsigned long bufferSpace, return false; } - if (!segment_pool.Init(numSegments, segmentSize + NormDataMsg::PayloadHeaderLength())) + if (!segment_pool.Init(numSegments, segmentSize + NormDataMsg::GetStreamPayloadHeaderLength() + 1)) { DMSG(0, "NormSession::StartServer() segment_pool init error\n"); StopServer(); @@ -213,7 +214,7 @@ bool NormSession::StartServer(unsigned long bufferSpace, if (numParity) { - if (!encoder.Init(numParity, segmentSize + NormDataMsg::PayloadHeaderLength())) + if (!encoder.Init(numParity, segmentSize + NormDataMsg::GetStreamPayloadHeaderLength())) { DMSG(0, "NormSession::StartServer() encoder init error\n"); StopServer(); @@ -231,6 +232,7 @@ bool NormSession::StartServer(unsigned long bufferSpace, is_server = true; flush_count = NORM_ROBUST_FACTOR+1; // (TBD) parameterize robust_factor //probe_timer.SetInterval(0.0); + probe_pending = probe_reset = false; OnProbeTimeout(probe_timer); ActivateTimer(probe_timer); return true; @@ -269,7 +271,6 @@ void NormSession::StopClient() if (!is_server) Close(); } - void NormSession::Serve() { // Only send new data when no other messages are queued for transmission @@ -876,7 +877,6 @@ void NormTrace(const struct timeval& currentTime, void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast) { - ASSERT(this == session_mgr.top_session); // Drop some rx messages for testing if (UniformRand(100.0) < rx_loss_rate) { @@ -1318,6 +1318,8 @@ void NormSession::ServerHandleAckMessage(const struct timeval& currentTime, cons void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, NormNackMsg& nack) { + // (TBD) maintain average of "numErasures" for SEGMENT repair requests + // to use as input to a an automatic "auto parity" adjustor // Update GRTT estimate struct timeval grttResponse; nack.GetGrttResponse(grttResponse); @@ -2090,6 +2092,7 @@ bool NormSession::SendMessage(NormMsg& msg) ProtoSystemTime(currentTime); bool clientMsg = false; + bool isProbe = false; // Fill in any last minute timestamps // (TBD) fill in SessionId fields on all messages as needed @@ -2104,7 +2107,8 @@ bool NormSession::SendMessage(NormMsg& msg) switch (((NormCmdMsg&)msg).GetFlavor()) { case NormCmdMsg::CC: - ((NormCmdCCMsg&)msg).SetSendTime(currentTime); + ((NormCmdCCMsg&)msg).SetSendTime(currentTime); + isProbe = true; break; default: break; @@ -2153,8 +2157,7 @@ bool NormSession::SendMessage(NormMsg& msg) } else { - if (tx_socket.SendTo(msg.GetBuffer(), - msgSize, + if (tx_socket.SendTo(msg.GetBuffer(), msgSize, msg.GetDestination())) { // Separate send/recv tracing @@ -2195,13 +2198,31 @@ bool NormSession::SendMessage(NormMsg& msg) result = false; } } + if (result && isProbe) + { + probe_pending = false; + if (probe_reset) + { + probe_reset = false; + OnProbeTimeout(probe_timer); + ActivateTimer(probe_timer); + } + } tx_timer.SetInterval(((double)msgSize) / tx_rate); return result; } // end NormSession::SendMessage() bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) { - // 1) Update grtt_estimate _if_ sufficient time elapsed. + // 1) Temporarily kill probe_timer if CMD(CC) not yet tx'd + if (probe_pending) + { + probe_reset = true; + probe_timer.Deactivate(); + return false; + } + + // 2) Update grtt_estimate _if_ sufficient time elapsed. grtt_age += probe_timer.GetInterval(); if (grtt_age >= grtt_interval) { @@ -2247,9 +2268,9 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) else grtt_interval *= 2.0; if (grtt_interval > grtt_interval_max) - grtt_interval = grtt_interval_max; + grtt_interval = grtt_interval_max; - // 2) Build probe message and determine next probe interval + // 3) Build a NORM_CMD(CC) message NormCmdCCMsg* cmd = (NormCmdCCMsg*)GetMessageFromPool(); if (!cmd) { @@ -2257,7 +2278,6 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) LocalNodeId()); return true; } - // Build a NORM_CMD(CC) message cmd->Init(); cmd->SetDestination(address); cmd->SetGrtt(grtt_quantized); @@ -2276,7 +2296,7 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) double probeInterval; if (cc_enable) { - // Iterate over cc_node_list and append + // Iterate over cc_node_list and append cc_nodes ... NormNodeListIterator iterator(cc_node_list); NormCCNode* next; while ((next = (NormCCNode*)iterator.GetNextNode())) @@ -2312,8 +2332,9 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) // Determine next probe_interval probeInterval = grtt_interval; } - - QueueMessage(cmd); + + QueueMessage(cmd); + probe_pending = true; // 3) Set probe_timer interval probe_timer.SetInterval(probeInterval); diff --git a/common/normSession.h b/common/normSession.h index c35e150..1f527bb 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -58,7 +58,7 @@ class NormSessionMgr ProtoTimerMgr& GetTimerMgr() {return timer_mgr;} ProtoSocket::Notifier& GetSocketNotifier() {return socket_notifier;} - private: + private: ProtoTimerMgr& timer_mgr; ProtoSocket::Notifier& socket_notifier; NormController* controller; @@ -320,6 +320,8 @@ class NormSession ProtoTimer probe_timer; // GRTT/congestion control probes bool probe_proactive; + bool probe_pending; // true while CMD(CC) enqueued + bool probe_reset; double grtt_interval; // current GRTT update interval double grtt_interval_min; // minimum GRTT update interval diff --git a/common/normSimAgent.cpp b/common/normSimAgent.cpp index 7f169bf..3047562 100644 --- a/common/normSimAgent.cpp +++ b/common/normSimAgent.cpp @@ -25,7 +25,7 @@ NormSimAgent::NormSimAgent(ProtoTimerMgr& timerMgr, // Bind NormSessionMgr to this agent and simulation environment session_mgr.SetController(static_cast(this)); - interval_timer.SetListener(this, (ProtoTimer::TimeoutHandler)&NormSimAgent::OnIntervalTimeout); + interval_timer.SetListener(this, &NormSimAgent::OnIntervalTimeout); interval_timer.SetInterval(0.0); interval_timer.SetRepeat(0); @@ -593,7 +593,7 @@ void NormSimAgent::Notify(NormController::Event event, } else { - OnIntervalTimeout(); + OnIntervalTimeout(interval_timer); } } break; @@ -790,7 +790,7 @@ void NormSimAgent::Notify(NormController::Event event, } // end NormSimAgent::Notify() -bool NormSimAgent::OnIntervalTimeout() +bool NormSimAgent::OnIntervalTimeout(ProtoTimer& theTimer) { if (tx_repeat_count) { diff --git a/common/normSimAgent.h b/common/normSimAgent.h index 2fa5e15..f48effc 100644 --- a/common/normSimAgent.h +++ b/common/normSimAgent.h @@ -42,7 +42,7 @@ class NormSimAgent : public NormController void ActivateTimer(ProtoTimer& theTimer) {session_mgr.ActivateTimer(theTimer);} - bool OnIntervalTimeout(); + bool OnIntervalTimeout(ProtoTimer& theTimer); static const char* const cmd_list[]; diff --git a/common/normVersion.h b/common/normVersion.h index 21ef689..ed93263 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.1b1" +#define VERSION "1.1b2" #endif // _NORM_VERSION diff --git a/common/raft.cpp b/common/raft.cpp new file mode 100644 index 0000000..64f9cdf --- /dev/null +++ b/common/raft.cpp @@ -0,0 +1,521 @@ + +#include "protokit.h" +#include // for sscanf() + +#ifdef UNIX +#include +#include +#endif // UNIX + + +class RaftApp : public ProtoApp +{ + public: + RaftApp(); + ~RaftApp(); + + // Overrides from ProtoApp base + bool OnStartup(int argc, const char*const* argv); + bool ProcessCommands(int argc, const char*const* argv); + void OnShutdown(); + + bool OnCommand(const char* cmd, const char* arg = NULL); + + private: + static void Usage() + { + fprintf(stderr, "Usage: raft [listen [/]][dest /]\n"); + } + enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG}; + static CmdType GetCommandType(const char* cmd); + static const char* const CMD_LIST[]; + + // Tunnel related members ... + void OnRxSocketEvent(ProtoSocket& theSocket, + ProtoSocket::Event theEvent); + static void DoInputReady(ProtoDispatcher::Descriptor descriptor, + ProtoDispatcher::Event event, + const void* userData); + void OnInputReady(); + + ProtoSocket rx_socket; + ProtoSocket tx_socket; + ProtoAddress tx_address; + char tx_msg_buffer[8194]; + UINT16 tx_msg_length; + UINT16 tx_msg_index; + + // RTSP proxy related members ... + void OnProxySocketEvent(ProtoSocket& theSocket, + ProtoSocket::Event theEvent); + void OnClientSocketEvent(ProtoSocket& theSocket, + ProtoSocket::Event theEvent); + + ProtoSocket rtsp_proxy_socket; + char* rtsp_url; + ProtoAddress rtsp_server_address; + ProtoSocket rtsp_client_socket; + +}; // end class RaftApp + +const char* const RaftApp::CMD_LIST[] = +{ + "+debug", // debug + "+listen", // recv [/] + "+dest", // send / + "+rtspProxy", // rtsp + NULL +}; + +RaftApp::RaftApp() + : rx_socket(ProtoSocket::UDP), + tx_socket(ProtoSocket::UDP), tx_msg_length(0), tx_msg_index(0), + rtsp_proxy_socket(ProtoSocket::TCP), rtsp_url(NULL), + rtsp_client_socket(ProtoSocket::TCP) +{ + rx_socket.SetNotifier(&GetSocketNotifier()); + rx_socket.SetListener(this, &RaftApp::OnRxSocketEvent); + + rtsp_proxy_socket.SetNotifier(&GetSocketNotifier()); + rtsp_proxy_socket.SetListener(this, &RaftApp::OnProxySocketEvent); + rtsp_client_socket.SetNotifier(&GetSocketNotifier()); + rtsp_client_socket.SetListener(this, &RaftApp::OnClientSocketEvent); +} + +RaftApp::~RaftApp() +{ + if (rtsp_proxy_socket.IsOpen()) + rtsp_proxy_socket.Close(); + if (rtsp_client_socket.IsOpen()) + rtsp_client_socket.Close(); + if (rtsp_url) + { + delete rtsp_url; + rtsp_url = NULL; + } +} + +bool RaftApp::OnStartup(int argc, const char*const* argv) +{ + bool result = ProcessCommands(argc, argv); + if (result && !dispatcher.IsPending()) + { + Usage(); + OnShutdown(); + return false; + } + return result; +} // end RaftApp::OnStartup() + +void RaftApp::OnShutdown() +{ + if (rtsp_proxy_socket.IsOpen()) + rtsp_proxy_socket.Close(); + if (rtsp_client_socket.IsOpen()) + rtsp_client_socket.Close(); + if (rtsp_url) + { + delete rtsp_url; + rtsp_url = NULL; + } +} // end RaftApp::OnShutdown() + +bool RaftApp::OnCommand(const char* cmd, const char* arg) +{ + if (!strncmp(cmd, "dest", strlen(cmd))) + { + char host[256]; + char* ptr = strchr(arg, '/'); + if (ptr) + { + unsigned int len = ptr - arg; + strncpy(host, arg, len); + host[len] = '\0'; + ptr++; + } + else + { + DMSG(0, "Raft::OnCommand() invalid \"dest\" command\n"); + return false; + } + if (!tx_address.ResolveFromString(host)) + { + DMSG(0, "Raft::OnCommand() invalid dest address\n"); + return false; + } + UINT16 port; + if (1 != sscanf(ptr, "%hu", &port)) + { + DMSG(0, "Raft::OnCommand() invalid dest port\n"); + return false; + } + tx_address.SetPort(port); + int fd = fileno(stdin); + if(-1 == fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK)) + { + DMSG(0, "Raft::OnCommand() warning: fcntl(stdout, F_SETFL(O_NONBLOCK)) error: %s", + strerror(errno)); + } + tx_msg_length = tx_msg_index = 0; + dispatcher.InstallGenericInput(fileno(stdin), RaftApp::DoInputReady, this); + } + else if (!strncmp(cmd, "listen", strlen(cmd))) + { + ProtoAddress groupAddr; + const char* ptr = strchr(arg, '/'); + if (ptr) + { + char group[256]; + unsigned int len = ptr - arg; + strncpy(group, arg, len); + group[len] = '\0'; + if (!groupAddr.ResolveFromString(group) || + !groupAddr.IsMulticast()) + { + DMSG(0, "Raft::OnCommand() invalid recv multicast address\n"); + return false; + } + ptr++; + } + else + { + ptr = arg; + } + UINT16 port; + if (1 != sscanf(ptr, "%hu", &port)) + { + DMSG(0, "Raft::OnCommand() invalid recv port\n"); + return false; + } + if (!rx_socket.Open(port)) + { + DMSG(0, "Raft::OnCommand() rx_socket.Open() error\n"); + return false; + } + if (groupAddr.IsValid()) + { + if (!rx_socket.JoinGroup(groupAddr)) + { + DMSG(0, "Raft::OnCommand() rx_socket.JoinGroup() error\n"); + return false; + } + } + } + else if (!strncmp(cmd, "rtspProxy", strlen(cmd))) + { + if (!rtsp_proxy_socket.Listen(554)) + { + DMSG(0, "RaftApp::OnCommand() error: rtsp_proxy_socket.Listen() failed\n"); + return false; + } + if (rtsp_url) delete rtsp_url; + if (!(rtsp_url = new char[strlen(arg) + 1])) + { + DMSG(0, "RaftApp::OnCommand() new rtsp_url error: %s\n", + strerror(errno)); + rtsp_proxy_socket.Close(); + return false; + } + strcpy(rtsp_url, arg); + char* ptr = strstr(rtsp_url, "rtsp://"); + if (!ptr) + { + DMSG(0, "RaftApp::OnCommand() error: invalid rtsp_url\n"); + rtsp_proxy_socket.Close(); + return false; + } + ptr += 7; + char hostName[256]; + if (1 != sscanf(ptr, "%255s", hostName)) + { + DMSG(0, "RaftApp::OnCommand() error: no rtsp_url hostname\n"); + rtsp_proxy_socket.Close(); + return false; + } + char* sptr = strchr(hostName, '/'); + if (sptr) *sptr = '\0'; + if (!rtsp_server_address.ResolveFromString(hostName)) + { + DMSG(0, "RaftApp::OnCommand() error: invalid rtsp_url hostname:%s\n", + hostName); + rtsp_proxy_socket.Close(); + return false; + } + rtsp_server_address.SetPort(554); + } + else + { + DMSG(0, "RaftApp::OnCommand() error: invalid command!\n"); + return false; + } + return true; +} // end RaftApp::OnCommand() + +bool RaftApp::ProcessCommands(int argc, const char*const* argv) +{ + int i = 1; + while (i < argc) + { + switch (GetCommandType(argv[i])) + { + case CMD_ARG: + if (!OnCommand(argv[i], argv[i+1])) + { + DMSG(0, "RaftApp::ProcessCommands() %s command error\n", argv[i]); + return false; + } + i += 2; + break; + case CMD_NOARG: + if (!OnCommand(argv[i])) + { + DMSG(0, "RaftApp::ProcessCommands() %s command error\n", argv[i]); + return false; + } + i++; + break; + default: + DMSG(0, "RaftApp::ProcessCommands() invalid command\n"); + return false; + } + } + return true; +} // end RaftApp::ProcessCommands() + +void RaftApp::DoInputReady(ProtoDispatcher::Descriptor /*descriptor*/, + ProtoDispatcher::Event /*event*/, + const void* userData) +{ + ((RaftApp*)userData)->OnInputReady(); +} // end RaftApp::DoInputReady() + +void RaftApp::OnInputReady() +{ + // Read from stdin ... + unsigned int want = tx_msg_length ? tx_msg_length - tx_msg_index : 2 - tx_msg_index; + if (want) + { + int result = fread(tx_msg_buffer+tx_msg_index, 1, want, stdin); + if (result > 0) + { + tx_msg_index += result; + } + else if (ferror(stdin)) + { + switch (errno) + { + case EINTR: + case EAGAIN: + break; + default: + DMSG(0, "raft: input error:%s\n", strerror(errno)); + break; + } + clearerr(stdin); + } + else if (feof(stdin)) + { + DMSG(0, "raft: input end-of-file\n"); + dispatcher.RemoveGenericInput(fileno(stdin)); + return; + } + } + if (0 == tx_msg_length) + { + if (2 == tx_msg_index) + { + memcpy(&tx_msg_length, tx_msg_buffer, 2); + tx_msg_length = ntohs(tx_msg_length); + if ((tx_msg_length < 2) || (tx_msg_length > 8194)) + { + DMSG(0, "raft: input error: invalid tx_msg_length: %u\n", tx_msg_length); + dispatcher.RemoveGenericInput(fileno(stdin)); + return; + } + } + OnInputReady(); + } + else if (tx_msg_index == tx_msg_length) + { + if (!tx_socket.SendTo(tx_msg_buffer+2, tx_msg_length-2, tx_address)) + { + DMSG(0, "raft: tx_socket.SendTo() error\n"); + return; + } + tx_msg_index = tx_msg_length = 0; + } + +} // end RaftApp::OnInputReady() + +void RaftApp::OnRxSocketEvent(ProtoSocket& /*theSocket*/, + ProtoSocket::Event theEvent) +{ + char buffer[8194]; + unsigned int numBytes = 8192; + ProtoAddress srcAddr; + if (!rx_socket.RecvFrom(buffer+2, numBytes, srcAddr)) + { + DMSG(0, "RaftApp::OnRxSocketEvent() rx_socket.RecvFrom() error\n"); + return; + } + numBytes += 2; + UINT16 msgLength = htons((UINT16)numBytes); + memcpy(buffer, &msgLength, 2); + unsigned int put = 0; + while (put < numBytes) + { + size_t result = fwrite(buffer+put, 1, numBytes-put, stdout); + if (result > 0) + { + put += result; + } + else if (EINTR != errno) + { + DMSG(0, "RaftApp::OnRxSocketEvent() fwrite() error: %s\n", + strerror(errno)); + } + } + fflush(stdout); +} // end RaftApp::OnRxSocketEvent() + +void RaftApp::OnProxySocketEvent(ProtoSocket& /*theSocket*/, + ProtoSocket::Event theEvent) +{ + //TRACE("RaftApp::OnProxySocketEvent() ...\n"); + switch (theEvent) + { + case ProtoSocket::ACCEPT: + { + if (!rtsp_proxy_socket.Accept()) + { + DMSG(0, "RaftApp::OnProxySocketEvent() rtsp_proxy_socket.Accept() error\n"); + rtsp_proxy_socket.Close(); + if (!rtsp_proxy_socket.Listen(554)) + DMSG(0, "raft: rtsp_proxy_socket.Listen() error\n"); + } + TRACE("calling rtsp_client_socket.Connect() ...\n"); + if (!rtsp_client_socket.Connect(rtsp_server_address)) + { + DMSG(0, "RaftApp::OnProxySocketEvent() rtsp_client_socket.Connect() error\n"); + rtsp_proxy_socket.Close(); + } + TRACE(" rtsp_client_socket.Connect() complete.\n"); + break; + } + case ProtoSocket::RECV: + { + //TRACE("rtsp_proxy_socket RECV event (connected:%d)...\n", + // rtsp_client_socket.IsConnected()); + if (rtsp_client_socket.IsConnected()) + { + char buffer[1024]; + unsigned int buflen = 1023; + while (rtsp_proxy_socket.Recv(buffer, buflen)) + { + unsigned int put = 0; + while (put < buflen) + { + unsigned int numBytes = buflen - put; + if (!rtsp_client_socket.Send(buffer+put, numBytes)) + { + DMSG(0, "rtsp_client_socket.Send() error\n"); + rtsp_client_socket.Close(); + rtsp_proxy_socket.Close(); + if (!rtsp_proxy_socket.Listen(554)) + DMSG(0, "raft: rtsp_proxy_socket.Listen() error\n"); + } + put += numBytes; + } + TRACE("rtsp_client sent %u bytes ....\n", put); + buflen = 1023; + } + TRACE("proxy socket RECV completed.\n"); + } + break; + } + case ProtoSocket::DISCONNECT: + { + TRACE("rtsp_proxy_socket DISCONNECT event ...\n"); + rtsp_proxy_socket.Close(); + rtsp_client_socket.Close(); + if (!rtsp_proxy_socket.Listen(554)) + DMSG(0, "raft: rtsp_proxy_socket.Listen() error\n"); + break; + } + default: + TRACE("rtsp_proxy_socket UNKNOWN event ...\n"); + break; + } +} // end RaftApp::OnProxySocketEvent() + +void RaftApp::OnClientSocketEvent(ProtoSocket& /*theSocket*/, + ProtoSocket::Event theEvent) +{ + TRACE("RaftApp::OnClientSocketEvent() ...\n"); + switch (theEvent) + { + case ProtoSocket::CONNECT: + { + TRACE("RaftApp::OnClientSocketEvent() CONNECT ...\n"); + break; + } + case ProtoSocket::RECV: + { + TRACE("rtsp_client_socket RECV event ..\n"); + char buffer[1024]; + unsigned int buflen = 1023; + while (rtsp_client_socket.Recv(buffer, buflen)) + { + buffer[buflen] = '\0'; + TRACE("%s"); + } + TRACE("\n"); + break; + } + case ProtoSocket::DISCONNECT: + { + TRACE("rtsp_client_socket DISCONNECT event ...\n"); + rtsp_client_socket.Close(); + rtsp_proxy_socket.Close(); + if (!rtsp_proxy_socket.Listen(554)) + DMSG(0, "raft: rtsp_proxy_socket.Listen() error\n"); + break; + } + default: + TRACE("rtsp_client_socket UNKNOWN event ...\n"); + break; + } +} // end RaftApp::OnClientSocketEvent() + +RaftApp::CmdType RaftApp::GetCommandType(const char* cmd) +{ + if (!cmd) return CMD_INVALID; + unsigned int len = strlen(cmd); + bool matched = false; + CmdType type = CMD_INVALID; + const char* const* nextCmd = CMD_LIST; + while (*nextCmd) + { + if (!strncmp(cmd, *nextCmd+1, len)) + { + if (matched) + { + // ambiguous command (command should match only once) + return CMD_INVALID; + } + else + { + matched = true; + if ('+' == *nextCmd[0]) + type = CMD_ARG; + else + type = CMD_NOARG; + } + } + nextCmd++; + } + return type; +} // end RaftApp::GetCommandType() + +PROTO_INSTANTIATE_APP(RaftApp) + diff --git a/unix/Makefile.common b/unix/Makefile.common index 9649a5b..23310bd 100644 --- a/unix/Makefile.common +++ b/unix/Makefile.common @@ -1,5 +1,5 @@ ######################################################################### -# COMMON MDP MAKEFILE STUFF +# COMMON NORM MAKEFILE STUFF # SHELL=/bin/sh @@ -40,13 +40,13 @@ TCL_SCRIPTS = ${TCL_SCRIPT_PATH}/init.tcl \ ${TK_SCRIPT_PATH}/text.tcl \ ${TK_SCRIPT_PATH}/optMenu.tcl -TARGETS = mdp tkMdp +TARGETS = norm raft # Rule for C++ .cpp extension .cpp.o: $(CC) -c $(CFLAGS) -o $*.o $*.cpp -# MDP depends upon the NRL Protean Group's development library +# NORM depends upon the NRL Protean Group's development library LIBPROTO = $(PROTOLIB)/unix/libProtokit.a $(PROTOLIB)/unix/libProtokit.a: cd $(PROTOLIB)/unix; $(MAKE) -f Makefile.$(SYSTEM) libProtokit.a @@ -76,13 +76,20 @@ libnormSim.a: $(SIM_OBJ) ar rcv $@ $(SIM_OBJ) ranlib $@ -# (mdp) command-line file broadcaster/receiver +# (norm) command-line file broadcaster/receiver APP_SRC = $(COMMON)/normApp.cpp $(COMMON)/normPostProcess.cpp \ $(UNIX)/unixPostProcess.cpp APP_OBJ = $(APP_SRC:.cpp=.o) norm: $(APP_OBJ) libnorm.a $(LIBPROTO) - $(CC) $(CFLAGS) -o $@ $(APP_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) + $(CC) $(CFLAGS) -o $@ $(APP_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) + +# (raft) command-line reliable tunnel helper +RAFT_SRC = $(COMMON)/raft.cpp +RAFT_OBJ = $(RAFT_SRC:.cpp=.o) + +raft: $(RAFT_OBJ) $(LIBPROTO) + $(CC) $(CFLAGS) -o $@ $(RAFT_OBJ) $(LDFLAGS) $(LIBPROTO) $(LIBS) clean: rm -f $(COMMON)/*.o $(UNIX)/*.o $(UNIX)/libnorm.a $(UNIX)/norm $(NS)/*.o; diff --git a/win32/Norm.opt b/win32/Norm.opt index 564b2386145f134072ee7643ca89b1f7e0e4bf82..055447b11c63ee02272e6cf76bb39719391b5f17 100644 GIT binary patch delta 2402 zcmZqp!`$$Pc|#5h+X2<{uU74yT*MN@!^XhCzyieofnc&AtIFgrEFm21$O0v-3nx2^ zici*I3kT7>dHF@TF(5)OrMO`8>JmG~&C{#8StjqBqC9zeyv}6%nC!_H>eMDTl&ejS zO;DXI7h5`ccdYW{f(q8jzav#9^CYWHJ}`-G@`gHv$sfv;fofSN|EgA)Twl&M`M^}g z$@-Htf%0l#^{dO3C(oOt1ZJ;~QJ(y-CJfB>t?-`ATc$GEFJ5)>{%LHJ&sVEYejlwe zIj@RsvPCk_WUXYW&4M|5SOgeZ7@3$DI2eA+UC$u!<2~bMMzIgflcYq4n~(H{n~yd* zaWGEqaL$=r;w&- z$%4KWpe&~f&0Z>#*Vm}Svmke}Iy|dEvl5vkJg6_%4Gv6b|a;$8?)1_4bAjg-VqC?o(p>f!r?>3!+Mbzn_Bm%}pPS~}XYcd3 zzn=d4AOC#jAGwL+YVK(6zmMi}$3x*WxF%A^b2-F6dPLX$nXVBe#}`Nf|G~2wsJEk_ zr$EO*ROcr^v!FRp9&{2!_53u>&w!onH9)f-a`Zee`px=Us&p&F&$Ql*&iv^wGgQ9Mw~f*Vxjj6;lWRf>CZ42@=V*dR7HNl(J2FFU zmD(;nM;brHAEAHLSaB?)bQ{tdqBN3Y*5f&pPko2_47dDKRESsp3ToZLyOIIfyUp%mA1qFLXR%N- zyJmA}A{}x?NWm!enyp={Z!WGbymi?q+57GUm-)af82MtuOWL5MOO8{tcRrj5-L!`; zUb1BlhZC_%0F%+NQua&QKuIKRylFmNynd=c%#`f?e0g9tU8~nk5!Xf(Y#T&tFs*6Ik@GJ8&=R^ME@e7@-RoI?J}?!wjTzU%C} z#AdUqx>xhmhADZ2B!C7=*XyXG zX2&d|*?L3V8+Mz^THX23Y`M&Q>t=@>=d)oZ=oUv5i!^CViCc9El+n$rK~ZysF4dpgy{_j!BZTtaaU3r z_j<)BZ7&$5d%ccPy77)ts$Miot45(%)RP(J4SlCa3EWnU!bW9f`PL16y;-YnRquoX zZZ6$kxwE=?`?8K#w@dZMAmtoOq?IPw~Jll0U zh$Y_k*d1P^WyB0|4$a7@M}K5g_e<^efMzgMcHGbyoepzi)-)A!XTOs@l{IvlZfh4q zLDm9m7`4px&H)o{HFq%1WG%C9xAx4$DIrpxTUPG?FSs=@c~7V&VcnR=g%UfwWt@k+ zE;br44%k&!Qo8EcJhbWm#+;mp#+vr{_^zX-e$lC5-9RJwE2B0T5gSM}E>5?kh*+<& zM8xR!5EVgHJ|3kLhU*A*z~2lLKa!=_u9C z5ObK5A11;|b=#l+ys0oW0g|*2WAY3AzNgPQw5z!3F2p(UDHE3>G`LQsi0z3P$5*eu z|KIVIpfWi`0wRGYR02mw8Xdy)1pgEKuOax~7+UCKBXQGx=m$bo@V`m0+()d&KT!n# z)4;mme?wF7KcgV{U!n@p8)7vmGs0dHzS<_fU-kEF_?tg&G>=ahpbeRljzEb_$pY*m zTPnf-NFhx+(=kPqs#G7^9-oPHl;D358Cz*3_+Ln^CHSANlvMCPf5~U)6LnlD zvBO)&^(nIc56es=kVYkKS^uY(TzL!r{AIx2?HInV!@_Me6cQlvI&1{B>-zvh_iknV zU*wW@S^p>N|Gr?Tg*>iEKqPP&5{R1ani~xsUe%~u!hw}eFkMe#0 z2;B02NZC)6zp3od$qsx^*@JVR;0((5WzQ}DMyULwu>V%3*mLD>b2z+hnfH%fS*g~p zhW|gif$zmND2Q9cV-9uPDW&-w3z{V(!%-fuY}hGeGE!VS^p{PKYhnuV*fqm`7ESd6Z>y2 zkRf~`0g=F`E`bj|(b(w=KE&rc=TeI^PI{r*1>%bd@ZE2d-i+h)J@5RTE@Cfu0~ zybPc2103D)Jm2lkz`1USnV!SrJE=YuT+(AFaLr==G4BqOSX6-ka+PPkgse`f0^3SK zpTPEvVrzP?!x9+v3$#<7?u2^}Zk>{zUqQro$?ac4Vb7|kLcO6yU@=AlrxHica|)Tj zr;6>9JIDkk1&iZHH#ve#V4K6!Iaa%y#G}_OkU?8QMyE%w=_(Ip3A{SPCcf1Zm~@8i z(K3t#-nkw4Q+0>!r2J-qLBl{MiDmBo(6O41mBiz&m5$6I`yI$CX~0!ghFRs$8zpnPAYTOTO*o~ z%Zb$9JqO>qXF6zQ0BzV{i(nhjaXAMccSHv^{ffm%PjrJ4*d#C zpSbl<7ycj8E~)uq40;>64j;wI`cGN^NzT|0(WSBYf6V6mIOMb`#dIOWIfdp$`;Hg< zk3Og65DAC`KJOCv*OA{HIf9Lx9$*)nYyW(ew$hP9Bp?zH35Wzl0wRGgxdeRu-#l&h zM%x(BPH#qSWo>1tzA{UQ$fWKsl5?IoCwD=(LFFRZ8mPIFvfI!6&GLSt724+447E)G1L%j z9YYojo~52)X$P|CXIRKMTMOb=wcrjo9*M_E=Uc z;&wyMxSXjU(72kmj)61K^9I7h8=AR|U-2(KqU6KkdNf-HsL>1z~|kbv#7; zzxW$8UlRSF=>HC2=PUca5fmVYNZ<*Tfaw38(2AGW772(1L;@m#Ct3o&{_jO!&lD)} z)MxoAKX_6L7{mXv&mUwHvH^^;{ENwlw183UF}lClVySLpilC^+qyyxr;}{l0EZ~|2 zC8ne4OpKNdx-%>b7yxHFlFCD%4@>!(MeV~>6lB9(Q%v{x7JelG86w_o?_8=j+Y;o52+N(n6zE1-!ZTY?w~n%SS%LRmQ4Amy~*Ld_}XK46{FNL^3gVWA*(*I|D#V~IYa^?fhSG^qW^p1s$5=JBp?zH z35Wzl0>S=A2Ngvgo;Nq@wzvzD_M_<;Yo$lUQ^uMD2Ra#Bb@s1qLj_Wr?|10`mqm;0v zVX@fI+E8dv^|poxQR}au4Nr`Q{ z149pGFCzFKZ4D*%A2r5eUIyxtz6zfA1uFR8V*ergzdMB( z+Y*I~YAdX%9i!eKA$A11k`<8Pe}p8+ArcS?eCiU&EA)?w^_TiRh3n(nO}Djs!N0iZ zUo7|+Z~6~Pd%160ee4+__+Q{0Snxl=|7zm@#WAOF4)OnjK|=h$I97^#AFfv<@&A%5 zkM;}-LE4Ah(ZZ_$L#6NlUs?Y*seO_4f3x$nmqM2AtJ|78FwJ7ZHc;aKMf|_`J4tL= zJ##SHSVH{2EEuJGy^c}30naJbi=owivi{GBMe{EoJ9g6fzPu9 zj$<3uT<$FPeHZ-Cpf8)@OfIjtW4$r3_Z`pRSbzVVx@k7s<{*_%@IS%-1phMy|BJz9 zU|i!^?QYU{Z2HPqCaNO-pZ!%(Oh8`jf5rYcYUtuS^G@%(g8vErR~$FmRR))JyR~P! zQ860Jibx{%zrmicp7ery!T<8`m)x~Gw%GqRdqcb5*r?Xl8{|UTGiz4Y;{W;c3{c3k Mi3CIfk4*yq1DRs9WdHyG diff --git a/win32/NormLib.dsp b/win32/NormLib.dsp index 8e29e91..1bc193d 100644 --- a/win32/NormLib.dsp +++ b/win32/NormLib.dsp @@ -63,8 +63,8 @@ LIB32=link.exe -lib # PROP Output_Dir "Debug" # PROP Intermediate_Dir "Debug" # PROP Target_Dir "" -# ADD BASE CPP /nologo /W3 /Gm /GX /ZI /Od /D "WIN32" /D "_DEBUG" /D "_MBCS" /D "_LIB" /YX /FD /GZ /c -# ADD CPP /nologo /W3 /Gm /GX /ZI /Od /I "..\common" /I "..\protolib\common" /I "..\protolib\win32" /D "_DEBUG" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" /D "WIN32" /D "_MBCS" /D "_LIB" /YX /FD /GZ /c +# ADD BASE CPP /nologo /W3 /Gm /GX /ZI /Od /D "WIN32" /D "_DEBUG" /D "_MBCS" /D "_LIB" /YX /FD /GZ /c +# ADD CPP /nologo /W3 /GX /Od /I "..\common" /I "..\protolib\common" /I "..\protolib\win32" /D "_DEBUG" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" /D "WIN32" /D "_MBCS" /D "_LIB" /YX /FD /GZ /c # ADD BASE RSC /l 0x409 /d "_DEBUG" # ADD RSC /l 0x409 /d "_DEBUG" BSC32=bscmake.exe @@ -89,11 +89,11 @@ SOURCE=..\common\galois.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -104,11 +104,11 @@ SOURCE=..\common\normBitmask.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -119,11 +119,11 @@ SOURCE=..\common\normEncoder.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -134,11 +134,11 @@ SOURCE=..\common\normFile.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -149,11 +149,11 @@ SOURCE=..\common\normMessage.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -164,11 +164,11 @@ SOURCE=..\common\normNode.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -179,11 +179,11 @@ SOURCE=..\common\normObject.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -194,11 +194,11 @@ SOURCE=..\common\normSegment.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF @@ -209,11 +209,11 @@ SOURCE=..\common\normSession.cpp !IF "$(CFG)" == "NormLib - Win32 Release" -# ADD CPP /MD /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MD /vmg /I "..\win32" !ELSEIF "$(CFG)" == "NormLib - Win32 Debug" -# ADD CPP /MDd /vmg /I "..\common" /I "..\win32" /I "..\protolib\common" /I "..\protolib\win32" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" +# ADD CPP /MDd /vmg /I "..\win32" !ENDIF diff --git a/win32/norm/norm.dsp b/win32/norm/norm.dsp index a76b1aa..ed644d7 100644 --- a/win32/norm/norm.dsp +++ b/win32/norm/norm.dsp @@ -69,7 +69,7 @@ LINK32=link.exe # PROP Ignore_Export_Lib 0 # PROP Target_Dir "" # ADD BASE CPP /nologo /W3 /Gm /GX /ZI /Od /D "WIN32" /D "_DEBUG" /D "_WINDOWS" /D "_MBCS" /YX /FD /GZ /c -# ADD CPP /nologo /MDd /W3 /Gm /vmg /GX /ZI /Od /I "..\..\common" /I "..\..\protolib\common" /I "..\..\protolib\win32" /D "_DEBUG" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" /D "WIN32" /D "_WINDOWS" /D "_MBCS" /YX /FD /GZ /c +# ADD CPP /nologo /MDd /W3 /vmg /GX /Od /I "..\..\common" /I "..\..\protolib\common" /I "..\..\protolib\win32" /D "_DEBUG" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" /D "WIN32" /D "_WINDOWS" /D "_MBCS" /YX /FD /GZ /c # ADD BASE MTL /nologo /D "_DEBUG" /mktyplib203 /win32 # ADD MTL /nologo /D "_DEBUG" /mktyplib203 /win32 # ADD BASE RSC /l 0x409 /d "_DEBUG"