From c74826c08f41ee8d2d39b5a0b954208c4cbfc08e Mon Sep 17 00:00:00 2001 From: Jeff Weston Date: Wed, 11 Sep 2019 11:31:13 -0400 Subject: [PATCH] v1.1b3 --- README.md | 6 +- VERSION.TXT | 7 + common/normApp.cpp | 7 +- common/normBitmask.cpp | 300 +++++++++++++++++++------------ common/normBitmask.h | 59 ++++-- common/normEncoder.cpp | 20 ++- common/normMessage.cpp | 85 --------- common/normMessage.h | 47 +++-- common/normNode.cpp | 160 +++++++++-------- common/normNode.h | 26 ++- common/normObject.cpp | 399 ++++++++++++++++++++++------------------- common/normObject.h | 45 ++++- common/normSegment.cpp | 142 ++++++++------- common/normSegment.h | 39 +++- common/normSession.cpp | 66 ++++--- common/normSession.h | 8 + common/normVersion.h | 2 +- ns/simplenorm.tcl | 4 +- 18 files changed, 800 insertions(+), 622 deletions(-) create mode 100644 VERSION.TXT diff --git a/README.md b/README.md index d3cc679..f6a2ff6 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,8 @@ 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. -I do have a PDF file which explains some of the command-line options. -Email me for it. - -We're getting there ;-) +The current code only has Makefiles for Unix platforms, but the +code could be assembled as Win32 project under VC++. The NORM code depends upon the current "Protolib" release. See for that code. diff --git a/VERSION.TXT b/VERSION.TXT new file mode 100644 index 0000000..5453ef8 --- /dev/null +++ b/VERSION.TXT @@ -0,0 +1,7 @@ + NORM Version History + +Version 1.1b3 +============= + - Started keeping version history. This release includes + some bug fixes, smoother congestion control operation, etc + over some of the earlier releases. diff --git a/common/normApp.cpp b/common/normApp.cpp index a1102be..06bb3fd 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -966,7 +966,10 @@ void NormApp::Notify(NormController::Event event, else { if (readLength > 0) - output_msg_sync = true; + output_msg_sync = true; + + TRACE("read %d bytes (index:%d) %hu...\n", readLength, output_index, + *((UINT16*)(output_buffer+output_index))); } if (readLength) @@ -1110,7 +1113,7 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/) if (tx_repeat_interval > tx_object_interval) { if (interval_timer.IsActive()) interval_timer.Deactivate(); - interval_timer.SetInterval(tx_repeat_interval = tx_object_interval); + interval_timer.SetInterval(tx_repeat_interval - tx_object_interval); ActivateTimer(interval_timer); return false; } diff --git a/common/normBitmask.cpp b/common/normBitmask.cpp index 17c2f4e..ba1040e 100644 --- a/common/normBitmask.cpp +++ b/common/normBitmask.cpp @@ -199,10 +199,10 @@ void NormBitmask::Destroy() } // end NormBitmask::Destroy() -unsigned long NormBitmask::NextSet(unsigned long index) const +bool NormBitmask::GetNextSet(unsigned long& index) const { - if (index >= num_bits) return num_bits; - if (index < first_set) return first_set; + if (index >= num_bits) return false; + if (index < first_set) return GetFirstSet(index); unsigned long maskIndex = index >> 3; if (mask[maskIndex]) { @@ -211,22 +211,28 @@ unsigned long NormBitmask::NextSet(unsigned long index) const for (int i = 0; i < w; i++) { int loc = BITLOCS[mask[maskIndex]][i]; - if (loc >= remainder) return (maskIndex << 3) + loc; + if (loc >= remainder) + { + index = (maskIndex << 3) + loc; + return true; + } } } while(++maskIndex < mask_len) { if (mask[maskIndex]) - return (maskIndex << 3) + - BITLOCS[mask[maskIndex]][0]; + { + index = (maskIndex << 3) + BITLOCS[mask[maskIndex]][0]; + return true; + } } - return num_bits; + return false; } // end NormBitmask::NextSet() -unsigned long NormBitmask::PrevSet(unsigned long index) const +bool NormBitmask::GetPrevSet(unsigned long& index) const { if (index >= num_bits) index = num_bits - 1; - if (index < first_set) return num_bits; + if (index < first_set) return false; unsigned long maskIndex = index >> 3; if (mask[maskIndex]) { @@ -237,7 +243,8 @@ unsigned long NormBitmask::PrevSet(unsigned long index) const int loc = BITLOCS[mask[maskIndex]][i]; if (loc <= remainder) { - return((maskIndex << 3) + loc); + index = ((maskIndex << 3) + loc); + return true; } } } @@ -248,26 +255,32 @@ unsigned long NormBitmask::PrevSet(unsigned long index) const if (mask[maskIndex]) { int w = WEIGHT[mask[maskIndex]] - 1; - return((maskIndex << 3) + BITLOCS[mask[maskIndex]][w]); + index = ((maskIndex << 3) + BITLOCS[mask[maskIndex]][w]); + return true; } } - return (num_bits); // (nothing was set) + return false; // (nothing prior was set) } // end NormBitmask::PrevSet() -unsigned long NormBitmask::NextUnset(unsigned long index) const +bool NormBitmask::GetNextUnset(unsigned long& index) const { - if (index >= num_bits) return num_bits; - unsigned long maskIndex = index >> 3; - unsigned char bit = 0x80 >> (index & 0x07); - while (index < num_bits) + if (index >= num_bits) return false; + unsigned long next = index; + unsigned long maskIndex = next >> 3; + unsigned char bit = 0x80 >> (next & 0x07); + while (next < num_bits) { unsigned char val = mask[maskIndex]; if (val) { - while (bit && (index < num_bits)) + while (bit && (next < num_bits)) { - if (0 == (val & bit)) return index; - index++; + if (0 == (val & bit)) + { + index = next; + return true; + } + next++; bit >>= 0x01; } bit = 0x80; @@ -275,10 +288,11 @@ unsigned long NormBitmask::NextUnset(unsigned long index) const } else { - return index; + index = next; + return true; } } - return index; + return false; } // end NormBitmask::NextUnset() @@ -337,8 +351,11 @@ bool NormBitmask::UnsetBits(unsigned long index, unsigned long count) count &= 0x07; if (count) mask[maskIndex+nbytes] &= 0xff >> count; } - if ((first_set >= index) && (end > first_set)) - first_set = NextSet(end); + if ((first_set >= index) && (end > first_set)) + { + first_set = end; + if (!GetNextSet(first_set)) first_set = num_bits; + } return true; } // end NormBitmask::UnsetBits() @@ -370,7 +387,9 @@ bool NormBitmask::Subtract(const NormBitmask& b) for(unsigned long i = 0; i < len; i++) mask[i] &= ~b.mask[i]; if (first_set >= b.first_set) - first_set = NextSet(first_set); + { + if (!GetNextSet(first_set)) first_set = num_bits; + } return true; } // end NormBitmask::Subtract() @@ -386,9 +405,14 @@ bool NormBitmask::XCopy(const NormBitmask& b) mask[i] = b.mask[i] & ~mask[i]; if (len < mask_len) memset(&mask[len], 0, mask_len - len); if (b.first_set < first_set) + { first_set = b.first_set; + } else - first_set = NextSet(b.first_set); + { + first_set = b.first_set; + GetNextSet(first_set); + } return true; } // end NormBitmask::XCopy() @@ -400,9 +424,14 @@ bool NormBitmask::Multiply(const NormBitmask& b) mask[i] |= b.mask[i]; if (len < mask_len) memset(&mask[len], 0, mask_len - len); if (b.first_set > first_set) - first_set = NextSet(b.first_set); + { + first_set = b.first_set; + if (!GetNextSet(first_set)) first_set = num_bits; + } else if (first_set > b.first_set) - first_set = NextSet(first_set); + { + if (!GetNextSet(first_set)) first_set = num_bits; + } return true; } // end NormBitmask::Multiply() @@ -413,7 +442,7 @@ bool NormBitmask::Xor(const NormBitmask& b) for(unsigned int i = 0; i < b.mask_len; i++) mask[i] ^= b.mask[i]; if (b.first_set == first_set) - first_set = NextSet(first_set); + GetNextSet(first_set); return true; } // end NormBitmask::Xor() @@ -445,7 +474,7 @@ NormSlidingMask::~NormSlidingMask() Destroy(); } -bool NormSlidingMask::Init(long numBits) +bool NormSlidingMask::Init(long numBits, UINT32 rangeMask) { if (mask) Destroy(); @@ -453,6 +482,8 @@ bool NormSlidingMask::Init(long numBits) unsigned long len = (numBits + 7) >> 3; if ((mask = new unsigned char[len])) { + range_mask = rangeMask; + range_sign = (rangeMask ^ (rangeMask >> 1)); mask_len = len; num_bits = numBits; Clear(); @@ -479,8 +510,8 @@ bool NormSlidingMask::CanSet(unsigned long index) const if (IsSet()) { // Determine position with respect to current start - // and end, given the "offset" of the current start - long pos = index - offset; + // and end, given the "offset" of the current start + long pos = Delta(index, offset); if (pos < 0) { // Precedes start. @@ -505,11 +536,12 @@ bool NormSlidingMask::CanSet(unsigned long index) const return true; } } + else if (pos < num_bits) + { + return true; + } else { - if (pos < num_bits) - return true; - else return false; // out of range } } @@ -526,12 +558,8 @@ bool NormSlidingMask::Set(unsigned long index) { // Determine position with respect to current start // and end, given the "offset" of the current start - long pos = index - offset; - if (pos <= -num_bits) - { - return false; // out-of-range - } - else if (pos < 0) + long pos = Delta(index , offset); + if (pos < 0) { // Precedes start. pos += start; @@ -557,11 +585,7 @@ bool NormSlidingMask::Set(unsigned long index) offset = index; } } - else if (pos >= num_bits) - { - return false; // out of range - } - else + else if (pos < num_bits) { pos += start; if (pos >= num_bits) pos -= num_bits; @@ -574,6 +598,10 @@ bool NormSlidingMask::Set(unsigned long index) if ((pos > end) || (pos < start)) end = pos; } } + else + { + return false; // out of range + } ASSERT((pos >> 3) >= 0); ASSERT((pos >> 3) < (long)mask_len); mask[(pos >> 3)] |= (0x80 >> (pos & 0x07)); @@ -592,11 +620,14 @@ bool NormSlidingMask::Unset(unsigned long index) //ASSERT(CanSet(index)); if (IsSet()) { - long pos = index - offset; - if (pos >= 0) + long pos = Delta(index, offset); + if (pos < 0) { - // Is it in range? - if (pos >= num_bits) return true; + return true; // out-of-range + } + else if (pos < num_bits) + { + // Is it in current range of set bits? pos += start; if (pos >= num_bits) pos -= num_bits; if (end < start) @@ -607,6 +638,7 @@ bool NormSlidingMask::Unset(unsigned long index) { if ((pos < start) || (pos > end)) return true; } + // Yes, it was in range. // Unset the corresponding bit ASSERT((pos >> 3) >= 0); ASSERT((pos >> 3) < (long)mask_len); @@ -619,29 +651,28 @@ bool NormSlidingMask::Unset(unsigned long index) } if (start == pos) { - unsigned long next = NextSet(index); - long delta = next - offset + start; - if (delta >= num_bits) - start = delta - num_bits; - else if (start < 0) - start = delta + num_bits; - else - start = delta; + unsigned long next = index; + if (!GetNextSet(next)) ASSERT(0); + long delta = Delta(next, offset); + ASSERT(delta >= 0); + start += delta; + if (start >= num_bits) start -= num_bits; offset = next; } - if (pos == end) { - unsigned long prev = PrevSet(index); - long delta = prev - offset + start; - if (delta >= num_bits) - end = delta - num_bits; - else if (delta < 0) - end = delta + num_bits; - else - end = delta; + unsigned long prev = index; + if (!GetPrevSet(prev)) ASSERT(0); + long delta = Delta(prev, offset); + ASSERT(delta >= 0); + end = start + delta; + if (end >= num_bits) end -= num_bits; } } + else + { + return true; // out-of-range + } } return true; } // end NormSlidingMask::Unset() @@ -658,8 +689,8 @@ bool NormSlidingMask::SetBits(unsigned long index, long count) long last = index + count - 1; if (!CanSet(index)) return false; if (!CanSet(last)) return false; - // Calculate first set bit position - firstPos = index - offset; + // Calculate first set bit position + firstPos = Delta(index, offset); if (firstPos < 0) { // precedes start @@ -674,7 +705,7 @@ bool NormSlidingMask::SetBits(unsigned long index, long count) if (firstPos >= num_bits) firstPos -= num_bits; } // Calculate last set bit position - lastPos = last - offset; + lastPos = Delta(last , offset); if (lastPos > 0) { // Is post start, post end? @@ -775,7 +806,7 @@ bool NormSlidingMask::UnsetBits(unsigned long index, long count) long firstPos; if (count <= 0) return true; if (count > num_bits) count = num_bits; - long delta = index - offset; + long delta = Delta(index , offset); if (delta >= num_bits) { return true; @@ -791,7 +822,9 @@ bool NormSlidingMask::UnsetBits(unsigned long index, long count) firstPos = start + delta; if (firstPos >= num_bits) firstPos -= num_bits; } - delta = index + count - 1 - LastSet(); + UINT32 lastSet; + if (!GetLastSet(lastSet)) ASSERT(0); + delta = Delta((index+count-1), lastSet); long lastPos; if (delta < 0) { @@ -863,17 +896,23 @@ bool NormSlidingMask::UnsetBits(unsigned long index, long count) mask[maskIndex+nbytes] &= 0xff >> count; } } - // Calling these will update the start/end state + // Calling these will properly update the offset/start/end state if (start == firstPos) { if (end == lastPos) + { start = end = num_bits; + } else - Unset(FirstSet()); + { + Unset(offset); + } } else if (end == lastPos) { - Unset(LastSet()); + UINT32 lastSet; + GetLastSet(lastSet); + Unset(lastSet); } } return true; @@ -883,7 +922,7 @@ bool NormSlidingMask::Test(unsigned long index) const { if (IsSet()) { - long pos = index - offset; + long pos = Delta(index , offset); if (pos >= 0) { // Is it in range? @@ -907,25 +946,26 @@ bool NormSlidingMask::Test(unsigned long index) const } // end NormSlidingMask::Test() -unsigned long NormSlidingMask::NextSet(unsigned long index) const +bool NormSlidingMask::GetNextSet(unsigned long& index) const { if (IsSet()) { - long pos = index - offset; + unsigned long next = index; + long pos = Delta(next, offset); if (pos >= 0) { // Is it in range? - if (pos >= num_bits) return (offset + num_bits); + if (pos >= num_bits) return false; pos += start; if (pos >= num_bits) pos -= num_bits; if (end < start) { - if ((pos > end) && (pos < start)) return (offset + num_bits); + if ((pos > end) && (pos < start)) return false; } else { - if ((pos < start) || (pos > end)) return (offset + num_bits); + if ((pos < start) || (pos > end)) return false; } // Seek next set bit unsigned long maskIndex = pos >> 3; @@ -941,7 +981,8 @@ unsigned long NormSlidingMask::NextSet(unsigned long index) const pos = (maskIndex << 3) + loc; pos -= start; if (pos < 0) pos += num_bits; - return (offset + pos); + index = offset + pos; + return true; } } } @@ -955,7 +996,8 @@ unsigned long NormSlidingMask::NextSet(unsigned long index) const pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][0]; pos -= start; if (pos < 0) pos += num_bits; - return (offset + pos); + index = offset + pos; + return true; } } maskIndex = 0; @@ -968,36 +1010,41 @@ unsigned long NormSlidingMask::NextSet(unsigned long index) const pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][0]; pos -= start; if (pos < 0) pos += num_bits; - return (offset + pos); + index = offset + pos; + return true; } } } else { - return offset; + return false; } } - return (offset + num_bits); // indicates nothing was set -} // end NormSlidingMask::NextSet() + return false; // indicates nothing was set +} // end NormSlidingMask::GetNextSet() -unsigned long NormSlidingMask::PrevSet(unsigned long index) const +bool NormSlidingMask::GetPrevSet(unsigned long& index) const { if (IsSet()) { - long pos = index - offset; + unsigned long prev = index; + long pos = Delta(prev, offset); if (pos >= 0) { // Is it in range? - if (pos >= num_bits) return (offset + num_bits); + if (pos >= num_bits) + { + return false; + } pos += start; if (pos >= num_bits) pos -= num_bits; if (end < start) { - if ((pos > end) && (pos < start)) return (offset + num_bits); + if ((pos > end) && (pos < start)) return false; } else { - if ((pos < start) || (pos > end)) return (offset + num_bits); + if ((pos < start) || (pos > end)) return false; } // Seek prev set bits, starting with index long maskIndex = pos >> 3; @@ -1013,7 +1060,8 @@ unsigned long NormSlidingMask::PrevSet(unsigned long index) const pos = (maskIndex << 3) + loc; pos -= start; if (pos < 0) pos += num_bits; - return (offset + pos); + index = offset + pos; + return true; } } } @@ -1029,7 +1077,8 @@ unsigned long NormSlidingMask::PrevSet(unsigned long index) const pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][w]; pos -= start; if (pos < 0) pos += num_bits; - return (offset + pos); + index = offset + pos; + return true; } } maskIndex = mask_len - 1; @@ -1043,15 +1092,16 @@ unsigned long NormSlidingMask::PrevSet(unsigned long index) const pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][w]; pos -= start; if (pos < 0) pos += num_bits; - return (offset + pos); + index = offset + pos; + return true; } } } } - return (offset + num_bits); // indicates nothing was set -} // end NormSlidingMask::PrevSet() + return false; // indicates nothing prior was set +} // end NormSlidingMask::GetPrevSet() -unsigned long NormSlidingMask::RawNextSet(const char* mask, long index, long end) +/*unsigned long NormSlidingMask::RawNextSet(const char* mask, long index, long end) { // Seek next set bit unsigned long maskIndex = index >> 3; @@ -1104,7 +1154,7 @@ unsigned long NormSlidingMask::RawPrevSet(const char* mask, long index, long sta } } return (index+1); // (nothing was set) -} // end NormSlidingMask::RawPrevSet() +} // end NormSlidingMask::RawPrevSet()*/ bool NormSlidingMask::Copy(const NormSlidingMask& b) { @@ -1115,7 +1165,11 @@ bool NormSlidingMask::Copy(const NormSlidingMask& b) if (range <= num_bits) { start = b.start & 0x07; - end = b.LastSet() - b.FirstSet() + start; + UINT32 bLastSet; + b.GetLastSet(bLastSet); + UINT32 bFirstSet; + b.GetFirstSet(bFirstSet); + end = bLastSet - bFirstSet + start; offset = b.offset; // Copy start to mask_len long startIndex = b.start >> 3; @@ -1166,11 +1220,16 @@ bool NormSlidingMask::Add(const NormSlidingMask& b) { if (IsSet()) { - if (!CanSet(b.FirstSet())) return false; - if (!CanSet(b.LastSet())) return false; + UINT32 bFirstSet; + b.GetFirstSet(bFirstSet); + if (!CanSet(bFirstSet)) return false; + UINT32 bLastSet; + b.GetFirstSet(bLastSet); + if (!CanSet(bLastSet)) return false; long range = b.end - b.start; if (range < 0) range += b.num_bits; - unsigned long index = b.FirstSet(); + unsigned long index; + b.GetFirstSet(index); for (long i = 0; i < range; i++) { // (TBD) Improve performance by getting/setting @@ -1206,10 +1265,6 @@ bool NormSlidingMask::Subtract(const NormSlidingMask& b) } } } - else - { - Copy(b); - } return true; } // end NormSlidingMask::Subtract() @@ -1221,9 +1276,14 @@ bool NormSlidingMask::XCopy(const NormSlidingMask& b) { if (IsSet()) { - if (!CanSet(b.FirstSet())) return false; - if (!CanSet(b.LastSet())) return false; - unsigned long index = b.FirstSet(); + UINT32 bFirstSet; + b.GetFirstSet(bFirstSet); + if (!CanSet(bFirstSet)) return false; + UINT32 bLastSet; + b.GetFirstSet(bLastSet); + if (!CanSet(bLastSet)) return false; + unsigned long index; + b.GetFirstSet(index); long range = b.end - b.start; if (range < 0) range += b.num_bits; for (long i = 0; i < range; i++) @@ -1237,9 +1297,13 @@ bool NormSlidingMask::XCopy(const NormSlidingMask& b) } else { - Copy(b); + return Copy(b); } } + else + { + Clear(); + } return true; } // end NormSlidingMask::XCopy() @@ -1268,13 +1332,19 @@ bool NormSlidingMask::Multiply(const NormSlidingMask& b) } // end NormSlidingMask::Multiply() // Logically XOR two bit mask such that "this = (this ^ b)" +// (TBD) rewrite for byte-based operation for better efficiency bool NormSlidingMask::Xor(const NormSlidingMask& b) { if (b.IsSet()) { - if (!CanSet(b.FirstSet())) return false; - if (!CanSet(b.LastSet())) return false; - unsigned long index = b.FirstSet(); + UINT32 bFirstSet; + b.GetFirstSet(bFirstSet); + if (!CanSet(bFirstSet)) return false; + UINT32 bLastSet; + b.GetFirstSet(bLastSet); + if (!CanSet(bLastSet)) return false; + unsigned long index; + b.GetFirstSet(index); long range = b.end - b.start; if (range < 0) range += b.num_bits; for (long i = 0; i < range; i++) @@ -1310,4 +1380,4 @@ void NormSlidingMask::Debug(long theCount) if (0x3f == (i & 0x3f)) DMSG(0, "\n "); } if (0x3f != (i & 0x3f)) DMSG(0, "\n"); -} // end NormSlidingMask::Display() +} // end NormSlidingMask::Debug() diff --git a/common/normBitmask.h b/common/normBitmask.h index a280d3e..a5351f4 100644 --- a/common/normBitmask.h +++ b/common/normBitmask.h @@ -37,11 +37,16 @@ class NormBitmask } bool IsSet() const {return (first_set < num_bits);} - unsigned long FirstSet() const {return first_set;} - unsigned long LastSet() const + bool GetFirstSet(unsigned long& index) const { - return ((first_set < num_bits) ? - PrevSet(num_bits - 1) : num_bits); + index = first_set; + return IsSet(); + } + + bool GetLastSet(unsigned long& index) const + { + index = num_bits - 1; + return GetPrevSet(index); } bool Test(unsigned long index) const { @@ -70,19 +75,19 @@ class NormBitmask if (index < num_bits) { mask[(index >> 3)] &= ~(0x80 >> (index & 0x07)); - (index == first_set) ? first_set = NextSet(index) : 0; + (index == first_set) ? first_set = GetNextSet(first_set) ? first_set : num_bits : 0; } return true; } - bool Invert(unsigned long index) + bool Invert(unsigned long index) {return (Test(index) ? Unset(index) : Set(index));} bool SetBits(unsigned long baseIndex, unsigned long count); bool UnsetBits(unsigned long baseIndex, unsigned long count); - unsigned long NextSet(unsigned long index) const; - unsigned long PrevSet(unsigned long index) const; - unsigned long NextUnset(unsigned long index) const; + bool GetNextSet(unsigned long& index) const; + bool GetPrevSet(unsigned long& index) const; + bool GetNextUnset(unsigned long& index) const; bool Copy(const NormBitmask &b); // this = b bool Add(const NormBitmask & b); // this = this | b @@ -120,7 +125,7 @@ class NormSlidingMask const char* GetMask() const {return (const char*)mask;} - bool Init(long numBits); + bool Init(long numBits, UINT32 rangeMax); void Destroy(); long Size() const {return num_bits;} void Clear() @@ -139,12 +144,17 @@ class NormSlidingMask offset = index; } bool IsSet() const {return (start < num_bits);} - unsigned long FirstSet() const {return offset;} - unsigned long LastSet() const + bool GetFirstSet(unsigned long& index) const + { + index = offset; + return IsSet(); + } + bool GetLastSet(unsigned long& index) const { long n = end - start; n = (n < 0) ? (n + num_bits) : n; - return (n + offset); + index = offset + n; + return IsSet(); } bool Test(unsigned long index) const; bool CanSet(unsigned long index) const; @@ -159,12 +169,12 @@ class NormSlidingMask - // These return "FirstSet()+Size()" when finding nothing - unsigned long NextSet(unsigned long index) const; - unsigned long PrevSet(unsigned long index) const; + // These return "false" when finding nothing + bool GetNextSet(unsigned long& index) const; + bool GetPrevSet(unsigned long& index) const; - static unsigned long RawNextSet(const char* mask, long index, long start); - static unsigned long RawPrevSet(const char* mask, long index, long end); + //static unsigned long RawNextSet(const char* mask, long index, long start); + //static unsigned long RawPrevSet(const char* mask, long index, long end); bool Copy(const NormSlidingMask& b); // this = b @@ -178,8 +188,21 @@ class NormSlidingMask void Debug(long theCount); private: + // Calculate "circular" delta between two indices + long Delta(unsigned long a, unsigned long b) const + { + long result = a - b; + return ((0 == (result & range_sign)) ? + (result & range_mask) : + (((result != range_sign) || (a < b)) ? + (result | ~range_mask) : result)); + } + + unsigned char* mask; unsigned long mask_len; + unsigned long range_mask; + long range_sign; long num_bits; long start; long end; diff --git a/common/normEncoder.cpp b/common/normEncoder.cpp index 3b3f5de..7dfd8ad 100644 --- a/common/normEncoder.cpp +++ b/common/normEncoder.cpp @@ -62,6 +62,11 @@ bool NormEncoder::Init(int numParity, int vecSizeMax) ASSERT(vecSizeMax >= 0); if (genPoly) Destroy(); npar = numParity; + +#ifdef SIMULATE + vecSizeMax = MIN(SIM_PAYLOAD_MAX+1, vecSizeMax); +#endif // SIMUATE + vector_size = vecSizeMax; // Create generator polynomial if(!CreateGeneratorPolynomial()) @@ -180,7 +185,7 @@ void NormEncoder::Encode(const char *data, char **pVec) // Copy pVec[0] for use in calculations #ifdef SIMULATE - UINT16 vecSize = MIN(SIM_PAYLOAD_MAX, vector_size); + UINT16 vecSize = MIN(SIM_PAYLOAD_MAX+1, vector_size); #else UINT16 vecSize = vector_size; #endif // if/else SIMULATE @@ -230,6 +235,10 @@ bool NormDecoder::Init(int numParity, int vecSizeMax) ASSERT((numParity>=0)&&(numParity<=128)); ASSERT(vecSizeMax >= 0); +#ifdef SIMULATE + vecSizeMax = MIN(SIM_PAYLOAD_MAX+1, vecSizeMax); +#endif // SIMUATE + if (Lambda) Destroy(); // Check if already inited ... npar = numParity; @@ -328,8 +337,7 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era // Debugging assertions ASSERT(Lambda); ASSERT(erasureCount && (erasureCount<=npar)); - - + // (A) Compute syndrome vectors // First zero out erasure vectors (MDP provides zero-filled vecs) @@ -337,7 +345,7 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era // Then calculate syndrome (based on zero value erasures) int nvecs = npar + ndata; #ifdef SIMULATE - int vecSize = MIN(SIM_PAYLOAD_MAX, vector_size); + int vecSize = MIN(SIM_PAYLOAD_MAX+1, vector_size); #else int vecSize = vector_size; #endif // if/else SIMUATE @@ -347,8 +355,8 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era { int X = gexp(i+1); unsigned char* synVec = sVec[i]; - memset(synVec, 0, vecSize*sizeof(char)); - for(int j = 0; j < nvecs; j++) + memset(synVec, 0, vecSize*sizeof(char)); + for(int j = 0; j < nvecs; j++) { unsigned char* data = dVec[j] ? (unsigned char*)dVec[j] : scratch; unsigned char* S = synVec; diff --git a/common/normMessage.cpp b/common/normMessage.cpp index 5176e50..81580db 100644 --- a/common/normMessage.cpp +++ b/common/normMessage.cpp @@ -425,30 +425,6 @@ unsigned char NormQuantizeRtt(double rtt) return ((unsigned char)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt))))); } // end NormQuantizeRtt() -bool NormObjectSize::operator<(const NormObjectSize& size) const -{ - UINT16 diff = msb - size.msb; - if (0 == diff) - return (lsb < size.lsb); - else if ((diff > 0x8000) || - ((0x8000 == diff) && (msb > size.msb))) - return true; - else - return false; -} // end NormObjectSize::operator<(NormObjectSize size) - -bool NormObjectSize::operator>(const NormObjectSize& size) const -{ - UINT16 diff = size.msb - msb; - if (0 == diff) - return (lsb > size.lsb); - else if ((diff > 0x8000) || - ((0x8000 == diff) && (size.msb > msb))) - return true; - else - return false; -} // end NormObjectSize::operator>(NormObjectSize id) - NormObjectSize NormObjectSize::operator+(const NormObjectSize& size) const { NormObjectSize total; @@ -557,64 +533,3 @@ NormObjectSize NormObjectSize::operator/(const NormObjectSize& b) const return result; } // end NormObjectSize::operator/(NormObjectSize b) - -bool NormObjectId::operator<(const NormObjectId& id) const -{ - UINT16 diff = value - id.value; - if ((diff > 0x8000) || - ((0x8000 == diff) && (value > id.value))) - return true; - else - return false; -} // end NormObjectId::operator<(NormObjectId id) - -bool NormObjectId::operator>(const NormObjectId& id) const -{ - UINT16 diff = id.value - value;; - if ((diff > 0x8000) || - ((0x8000 == diff) && (id.value > value))) - return true; - else - return false; -} // end NormObjectId::operator>(NormObjectId id) - -int NormObjectId::operator-(const NormObjectId& id) const -{ - UINT16 diff = value - id.value; - if ((diff > 0x8000) || - ((0x8000 == diff) && (value > id.value))) - return ((INT16)(value - id.value)); - else - return (-(INT16)(id.value - value)); -} // end NormObjectId::operator-(NormObjectId id) const - - -bool NormBlockId::operator<(const NormBlockId& id) const -{ - UINT32 diff = value - id.value; - if ((diff > 0x80000000) || - ((0x80000000 == diff) && (value > id.value))) - return true; - else - return false; -} // end NormBlockId::operator<(NormBlockId id) - -bool NormBlockId::operator>(const NormBlockId& id) const -{ - UINT32 diff = id.value - value;; - if ((diff > 0x80000000) || - ((0x80000000 == diff) && (id.value > value))) - return true; - else - return false; -} // end NormBlockId::operator>(NormBlockId id) - -long NormBlockId::operator-(const NormBlockId& id) const -{ - UINT32 diff = value - id.value; - if ((diff > 0x80000000) || - ((0x80000000 == diff) && (value > id.value))) - return ((INT32)(value - id.value)); - else - return (-(INT32)(id.value - value)); -} // end NormBlockId::operator-(NormBlockId id) const diff --git a/common/normMessage.h b/common/normMessage.h index f648fe0..e99314c 100644 --- a/common/normMessage.h +++ b/common/normMessage.h @@ -106,15 +106,24 @@ class NormObjectSize UINT16 MSB() const {return msb;} UINT32 LSB() const {return lsb;} - - bool operator<(const NormObjectSize& size) const; - bool operator<=(const NormObjectSize& size) const - {return ((*this == size) || (*this(const NormObjectSize& size) const; - bool operator>=(const NormObjectSize& size) const - {return ((*this == size) || (*this>size));} bool operator==(const NormObjectSize& size) const {return ((msb == size.msb) && (lsb == size.lsb));} + bool operator>(const NormObjectSize& size) const + { + return ((msb == size.msb) ? + (lsb > size.lsb) : + (msb > size.msb)); + } + bool operator<(const NormObjectSize& size) const + { + return ((msb == size.msb) ? + (lsb < size.lsb) : + (msb < size.msb)); + } + bool operator<=(const NormObjectSize& size) const + {return ((*this == size) || (*this=(const NormObjectSize& size) const + {return ((*this == size) || (*this>size));} bool operator!=(const NormObjectSize& size) const {return ((lsb != size.lsb) || (msb != size.msb));} NormObjectSize operator+(const NormObjectSize& size) const; @@ -158,17 +167,26 @@ class NormObjectId NormObjectId(UINT16 id) {value = id;} NormObjectId(const NormObjectId& id) {value = id.value;} operator UINT16() const {return value;} - bool operator<(const NormObjectId& id) const; + int operator-(const NormObjectId& id) const + { + int result = value - id.value; + return ((0 == (result & 0x00008000)) ? + (result & 0x0000ffff) : + (((result != 0x00008000) || (value < id.value)) ? + result | 0xffff0000 : result)); + } + bool operator<(const NormObjectId& id) const + {return ((*this - id) < 0);} + bool operator>(const NormObjectId& id) const + {return ((*this - id) > 0);} bool operator<=(const NormObjectId& id) const {return ((value == id.value) || (*this(const NormObjectId& id) const; bool operator>=(const NormObjectId& id) const {return ((value == id.value) || (*this>id));} bool operator==(const NormObjectId& id) const {return (value == id.value);} bool operator!=(const NormObjectId& id) const {return (value != id.value);} - int operator-(const NormObjectId& id) const; NormObjectId& operator++(int ) {value++; return *this;} private: @@ -182,13 +200,16 @@ class NormBlockId NormBlockId(UINT32 id) {value = id;} NormBlockId(const NormObjectId& id) {value = (UINT32)id;} operator UINT32() const {return value;} - bool operator<(const NormBlockId& id) const; - bool operator>(const NormBlockId& id) const; bool operator==(const NormBlockId& id) const {return (value == (UINT32)id);} bool operator!=(const NormBlockId& id) const {return (value != (UINT32)id);} - long operator-(const NormBlockId& id) const; + long operator-(const NormBlockId& id) const + {return ((long)value - id.value);} + bool operator<(const NormBlockId& id) const + {return ((*this - id) < 0);} + bool operator>(const NormBlockId& id) const + {return ((*this - id) > 0);} NormBlockId& operator++(int ) {value++; return *this;} private: diff --git a/common/normNode.cpp b/common/normNode.cpp index 79b1546..8b3410c 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -82,13 +82,13 @@ bool NormServerNode::Open(UINT16 sessionId, UINT16 segmentSize, UINT16 numData, Close(); return false; } - if (!rx_pending_mask.Init(256)) + if (!rx_pending_mask.Init(256, 0x0000ffff)) { DMSG(0, "NormServerNode::Open() rx_pending_mask init error\n"); Close(); return false; } - if (!rx_repair_mask.Init(256)) + if (!rx_repair_mask.Init(256, 0x0000ffff)) { DMSG(0, "NormServerNode::Open() rx_repair_mask init error\n"); Close(); @@ -271,7 +271,7 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime, } else if (!cc_timer.IsActive()) { - double backoffFactor = session->BackoffFactor(); + double backoffFactor = backoff_factor; backoffFactor = MAX(backoffFactor, 4.0); double maxBackoff = grtt_estimate*backoffFactor; // (TBD) don't backoff for unicast sessions @@ -312,6 +312,7 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime, case NormCmdMsg::FLUSH: // (TBD) should we force synchronize if we're expected // to positively acknowledge the FLUSH + if (synchronized) { const NormCmdFlushMsg& flush = (const NormCmdFlushMsg&)cmd; @@ -397,7 +398,7 @@ void NormServerNode::HandleCCFeedback(UINT8 ccFlags, double ccRate) if (suppressed) { if (cc_timer.IsActive()) cc_timer.Deactivate(); - cc_timer.SetInterval(grtt_estimate*session->BackoffFactor()); + cc_timer.SetInterval(grtt_estimate*backoff_factor); // (TBD) ??? session->ActivateTimer(cc_timer); cc_timer.DecrementRepeatCount(); } @@ -910,11 +911,12 @@ void NormServerNode::Sync(NormObjectId objectId) { if (synchronized) { - if (rx_pending_mask.IsSet()) + NormObjectId firstPending; + if (GetFirstPending(firstPending)) { - NormObjectId firstSet = NormObjectId((UINT16)rx_pending_mask.FirstSet()); - if ((objectId > NormObjectId((UINT16)rx_pending_mask.LastSet())) || - ((next_id - objectId) > 256)) + NormObjectId lastPending; + GetLastPending(lastPending); + if ((objectId > lastPending) || ((next_id - objectId) > 256)) { NormObject* obj; while ((obj = rx_table.Find(rx_table.RangeLo()))) @@ -924,7 +926,7 @@ void NormServerNode::Sync(NormObjectId objectId) } rx_pending_mask.Clear(); } - else if (objectId > firstSet) + else if (objectId > firstPending) { NormObject* obj; while ((obj = rx_table.Find(rx_table.RangeLo())) && @@ -933,12 +935,11 @@ void NormServerNode::Sync(NormObjectId objectId) DeleteObject(obj); failure_count++; } - unsigned long numBits = (UINT16)(objectId - firstSet) + 1; - rx_pending_mask.UnsetBits(firstSet, numBits); + unsigned long numBits = (UINT16)(objectId - firstPending) + 1; + rx_pending_mask.UnsetBits(firstPending, numBits); } } - if ((next_id < objectId) || - ((next_id - objectId) > 256)) + if ((next_id < objectId) || ((next_id - objectId) > 256)) { max_pending_object = next_id = objectId; } @@ -963,7 +964,7 @@ NormServerNode::ObjectStatus NormServerNode::UpdateSyncStatus(const NormObjectId // (TBD) We may want to control re-sync policy options // or revert to fresh sync if sync is totally lost, // otherwise SQUELCH process will get things in order - DMSG(2, "NormServerNode::HandleObjectMessage() node>%lu re-syncing to server>%lu...\n", + DMSG(2, "NormServerNode::UpdateSyncStatus() node>%lu re-syncing to server>%lu...\n", LocalNodeId(), GetId()); Sync(objectId); resync_count++; @@ -990,7 +991,7 @@ void NormServerNode::SetPending(NormObjectId objectId) rx_pending_mask.SetBits(next_id, objectId - next_id + 1); next_id = objectId + 1; // This prevents the "sync_id" from getting stale - sync_id = (UINT16)rx_pending_mask.FirstSet(); + GetFirstPending(sync_id); } } // end NormServerNode::SetPending() @@ -1068,19 +1069,18 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel, if (!repair_timer.IsActive()) { bool startTimer = false; - if (rx_pending_mask.IsSet()) + NormObjectId nextId; + if (GetFirstPending(nextId)) { if (rx_repair_mask.IsSet()) rx_repair_mask.Clear(); - NormObjectId nextId = (UINT16)rx_pending_mask.FirstSet(); - NormObjectId lastId = (UINT16)rx_pending_mask.LastSet(); - if (objectId < lastId) lastId = objectId; - while (nextId <= lastId) + do { + if (nextId > objectId) break; NormObject* obj = rx_table.Find(nextId); if (obj) { NormObject::CheckLevel level; - if (nextId < lastId) + if (nextId < objectId) level = NormObject::THRU_OBJECT; else level = checkLevel; @@ -1092,18 +1092,17 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel, startTimer = true; } nextId++; - nextId = (UINT16)rx_pending_mask.NextSet(nextId); - } + } while (GetNextPending(nextId)); current_object_id = objectId; if (startTimer) { // BACKOFF related code double backoffInterval = - (session->Address().IsMulticast() && (session->BackoffFactor() > 0.0)) ? - ExponentialRand(grtt_estimate*session->BackoffFactor(), gsize_estimate) : + (session->Address().IsMulticast() && (backoff_factor > 0.0)) ? + ExponentialRand(grtt_estimate*backoff_factor, gsize_estimate) : 0.0; repair_timer.SetInterval(backoffInterval); - DMSG(3, "NormServerNode::RepairCheck() node>%lu begin NACK back-off: %lf sec)...\n", + DMSG(4, "NormServerNode::RepairCheck() node>%lu begin NACK back-off: %lf sec)...\n", LocalNodeId(), backoffInterval); session->ActivateTimer(repair_timer); } @@ -1155,14 +1154,15 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu end NACK back-off ...\n", LocalNodeId()); // 1) Were we suppressed? - if (rx_pending_mask.IsSet()) + NormObjectId nextId; + if (GetFirstPending(nextId)) { + // This loop checks to see if we have any repair pending objects + // (If we don't have any, that means we were suppressed) bool repairPending = false; - NormObjectId nextId = (UINT16)rx_pending_mask.FirstSet(); - NormObjectId lastId = (UINT16)rx_pending_mask.LastSet(); - if (current_object_id < lastId) lastId = current_object_id; - while (nextId <= lastId) + do { + if (nextId > current_object_id) break; if (!rx_repair_mask.Test(nextId)) { NormObject* obj = rx_table.Find(nextId); @@ -1173,8 +1173,8 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) } } nextId++; - nextId = (UINT16)rx_pending_mask.NextSet(nextId); - } // end while (nextId <= current_block_id) + } while (GetNextPending(nextId)); + if (repairPending) { // We weren't completely suppressed, so build NACK @@ -1222,38 +1222,34 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) if (cc_timer.IsActive()) { cc_timer.Deactivate(); - cc_timer.SetInterval(grtt_estimate*session->BackoffFactor()); + cc_timer.SetInterval(grtt_estimate*backoff_factor); session->ActivateTimer(cc_timer); cc_timer.DecrementRepeatCount(); } } + // Iterate through rx pending object list, appending repair requests as needed NormRepairRequest req; - NormObjectId prevId; - UINT16 reqCount = 0; NormRepairRequest::Form prevForm = NormRepairRequest::INVALID; - nextId = (UINT16)rx_pending_mask.FirstSet(); - lastId = (UINT16)rx_pending_mask.LastSet(); - if (max_pending_object < lastId) lastId = max_pending_object; - lastId++; // force loop to fully flush nack building. - while ((nextId <= lastId) || (reqCount > 0)) + bool iterating = GetFirstPending(nextId); + iterating = iterating && (nextId <= max_pending_object); + NormObjectId prevId = nextId; + UINT16 consecutiveCount = 0; + while (iterating || (0 != consecutiveCount)) { - NormObject* obj = NULL; - bool objPending = false; - if (nextId == lastId) - nextId++; // force break of possible ending consecutive series + bool appendRequest = false; + NormObject* obj = iterating ? rx_table.Find(nextId) : NULL; + if (obj) + appendRequest = true; + else if (iterating && ((nextId - prevId) == consecutiveCount)) + consecutiveCount++; // consecutive series of missing objs starts/continues else - obj = rx_table.Find(nextId); - if (obj) objPending = obj->IsPending(nextId != max_pending_object); + appendRequest = true; // consecutive series broken or finished - if (!objPending && reqCount && (reqCount == (nextId - prevId))) - { - reqCount++; // consecutive series of missing objs continues - } - else + if (appendRequest) { NormRepairRequest::Form nextForm; - switch (reqCount) + switch (consecutiveCount) { case 0: nextForm = NormRepairRequest::INVALID; @@ -1265,7 +1261,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) default: nextForm = NormRepairRequest::RANGES; break; - } + } if (prevForm != nextForm) { if (NormRepairRequest::INVALID != prevForm) @@ -1280,41 +1276,47 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) } prevForm = nextForm; } - if (NormRepairRequest::INVALID != nextForm) - DMSG(6, "NormServerNode::AppendRepairRequest() OBJECT request\n"); switch (nextForm) { case NormRepairRequest::ITEMS: req.AppendRepairItem(prevId, 0, ndata, 0); - if (2 == reqCount) + if (2 == consecutiveCount) req.AppendRepairItem(prevId+1, 0, ndata, 0); break; case NormRepairRequest::RANGES: - req.AppendRepairItem(prevId, 0, ndata, 0); - req.AppendRepairItem(prevId+reqCount-1, 0, ndata, 0); + req.AppendRepairRange(prevId, 0, ndata, 0, + prevId+consecutiveCount-1, 0, ndata, 0); break; default: break; } - prevId = nextId; - if (obj || (nextId >= lastId)) - reqCount = 0; + + if (obj) + { + if (obj->IsPending(nextId != max_pending_object)) + { + if (NormRepairRequest::INVALID != prevForm) + nack->PackRepairRequest(req); // (TBD) error check + prevForm = NormRepairRequest::INVALID; + bool flush = (nextId != max_pending_object); + obj->AppendRepairRequest(*nack, flush); // (TBD) error check + } + consecutiveCount = 0; + } + else if (iterating) + { + consecutiveCount = 1; + } else - reqCount = 1; - } // end if/else (!objPending && reqCount && (reqCount == (nextId - prevId))) - if (objPending) - { - if (NormRepairRequest::INVALID != prevForm) - nack->PackRepairRequest(req); // (TBD) error check - prevForm = NormRepairRequest::INVALID; - reqCount = 0; - bool flush = (nextId != max_pending_object); - obj->AppendRepairRequest(*nack, flush); // (TBD) error check - } + { + consecutiveCount = 0; // we're all done + } + prevId = nextId; + } // end if (appendRequest) nextId++; - if (nextId <= lastId) - nextId = (UINT16)rx_pending_mask.NextSet(nextId); - } // end while(nextId <= lastId) + iterating = GetNextPending(nextId); + } while (iterating || (0 != consecutiveCount)); + // Now that all repair requests have been appended, pack the nack if (NormRepairRequest::INVALID != prevForm) nack->PackRepairRequest(req); // (TBD) error check // Queue NACK for transmission @@ -1338,9 +1340,9 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) } // end if/else(repairPending) // BACKOFF related code double holdoffInterval = - session->Address().IsMulticast() ? grtt_estimate*(session->BackoffFactor() + 2.0) : + session->Address().IsMulticast() ? grtt_estimate*(backoff_factor + 2.0) : grtt_estimate; - holdoffInterval = (session->BackoffFactor() > 0.0) ? holdoffInterval : 1.01*grtt_estimate; + holdoffInterval = (backoff_factor > 0.0) ? holdoffInterval : 1.01*grtt_estimate; repair_timer.SetInterval(holdoffInterval); DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu begin NACK hold-off: %lf sec ...\n", @@ -1514,7 +1516,7 @@ bool NormServerNode::OnCCTimeout(ProtoTimer& /*theTimer*/) } // Begin cc_timer "holdoff" phase - cc_timer.SetInterval(grtt_estimate*session->BackoffFactor()); + cc_timer.SetInterval(grtt_estimate*backoff_factor); return true; } diff --git a/common/normNode.h b/common/normNode.h index 0f0d48d..d1a5832 100644 --- a/common/normNode.h +++ b/common/normNode.h @@ -209,9 +209,31 @@ class NormServerNode : public NormNode bool SyncTest(const NormObjectMsg& msg) const; void Sync(NormObjectId objectId); ObjectStatus UpdateSyncStatus(const NormObjectId& objectId); - void SetPending(NormObjectId objectId); ObjectStatus GetObjectStatus(const NormObjectId& objectId) const; + bool GetFirstPending(NormObjectId& objectId) + { + UINT32 index; + bool result = rx_pending_mask.GetFirstSet(index); + objectId = (UINT16)index; + return result; + } + bool GetNextPending(NormObjectId& objectId) + { + UINT32 index = (UINT16)objectId; + bool result = rx_pending_mask.GetNextSet(index); + objectId = (UINT16)index; + return result; + } + bool GetLastPending(NormObjectId& objectId) + { + UINT32 index; + bool result = rx_pending_mask.GetLastSet(index); + objectId = (UINT16)index; + return result; + } + void SetPending(NormObjectId objectId); + void DeleteObject(NormObject* obj); UINT16 SegmentSize() {return segment_size;} @@ -237,7 +259,7 @@ class NormServerNode : public NormNode {return erasure_loc[index];} UINT16 Decode(char** segmentList, UINT16 numData, UINT16 erasureCount) { - return decoder.Decode(segmentList, numData, erasureCount, erasure_loc); + return decoder.Decode(segmentList, numData, erasureCount, erasure_loc); } void CalculateGrttResponse(const struct timeval& currentTime, diff --git a/common/normObject.cpp b/common/normObject.cpp index 912a6d9..007291a 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -102,7 +102,7 @@ bool NormObject::Open(const NormObjectSize& objectSize, } // Init pending_mask (everything pending) - if (!pending_mask.Init(numBlocks.LSB())) + if (!pending_mask.Init(numBlocks.LSB(), 0xffffffff)) { DMSG(0, "NormObject::Open() init pending_mask error\n"); Close(); @@ -112,7 +112,7 @@ bool NormObject::Open(const NormObjectSize& objectSize, pending_mask.SetBits(0, numBlocks.LSB()); // Init repair_mask - if (!repair_mask.Init(numBlocks.LSB())) + if (!repair_mask.Init(numBlocks.LSB(), 0xffffffff)) { DMSG(0, "NormObject::Open() init pending_mask error\n"); Close(); @@ -152,8 +152,8 @@ bool NormObject::Open(const NormObjectSize& objectSize, ASSERT(0 == smallBlockCount.MSB()); small_block_count = smallBlockCount.LSB(); } - - final_block_id = large_block_count + small_block_count - 1; // not used for STREAM objects + // not used for STREAM objects + final_block_id = large_block_count + small_block_count - 1; NormObjectSize finalSegmentSize = objectSize - (numSegments - NormObjectSize((UINT32)1))*segmentSize; ASSERT(0 == finalSegmentSize.MSB()); @@ -291,26 +291,24 @@ bool NormObject::ActivateRepairs() repairsActivated = true; } // Activate any complete block repairs - if (repair_mask.IsSet()) + NormBlockId nextId; + if (GetFirstRepair(nextId)) { + NormBlockId lastId; + ASSERT(GetLastRepair(lastId)); + DMSG(6, "NormObject::ActivateRepairs() node>%lu obj>%hu activating blk>%lu->%lu repairs\n", + LocalNodeId(), (UINT16)id, (UINT32)nextId, (UINT32)lastId); repairsActivated = true; - NormBlockId nextId = repair_mask.FirstSet(); - NormBlockId lastId = repair_mask.LastSet(); - DMSG(6, "NormObject::ActivateRepairs() node>%lu obj>%hu activated blk>%lu->%lu repairs\n", - LocalNodeId(), (UINT16)id, - (UINT32)nextId, (UINT32)lastId); UINT16 autoParity = session->ServerAutoParity(); - while (nextId <= lastId) + do { NormBlock* block = block_buffer.Find(nextId); if (block) block->TxReset(GetBlockSize(nextId), nparity, autoParity, segment_size); // (TBD) This check can be eventually eliminated if everything else is done right if (!pending_mask.Set(nextId)) - DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error!\n", - (UINT32)nextId); - nextId++; - nextId = repair_mask.NextSet(nextId); - } + DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error!\n", (UINT32)nextId); + nextId++; + } while (GetNextRepair(nextId)); repair_mask.Clear(); } // Activate partial block (segment) repairs @@ -335,8 +333,10 @@ bool NormObject::ActivateRepairs() bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd) { // Determine range of blocks possibly pending repair - NormBlockId nextId = repair_mask.FirstSet(); - NormBlockId endId = repair_mask.LastSet(); + NormBlockId nextId; + GetFirstRepair(nextId); + NormBlockId endId; + GetLastRepair(endId); if (block_buffer.IsEmpty()) { if (repair_mask.IsSet()) endId++; @@ -466,9 +466,9 @@ bool NormObject::IsPending(bool flush) const } else { - if (pending_mask.IsSet()) + NormBlockId firstId; + if (GetFirstPending(firstId)) { - NormBlockId firstId = pending_mask.FirstSet(); if (firstId < max_pending_block) { return true; @@ -484,7 +484,10 @@ bool NormObject::IsPending(bool flush) const NormBlock* block = block_buffer.Find(max_pending_block); if (block) { - if (block->FirstPending() < max_pending_segment) + ASSERT(block->IsPending()); + NormSegmentId firstPending = 0; + block->GetFirstPending(firstPending); + if (firstPending < max_pending_segment) return true; else return false; @@ -552,6 +555,7 @@ bool NormObject::ClientRepairCheck(CheckLevel level, else if (blockId > max_pending_block) max_pending_block = blockId; max_pending_segment = GetBlockSize(max_pending_block); + break; default: break; } // end switch (level) @@ -605,37 +609,43 @@ bool NormObject::ClientRepairCheck(CheckLevel level, if ((level > THRU_INFO) && pending_mask.IsSet()) { if (repair_mask.IsSet()) repair_mask.Clear(); - NormBlockId nextId = pending_mask.FirstSet(); - NormBlockId lastId = pending_mask.LastSet(); - if ((level < THRU_OBJECT) && (blockId < lastId)) - lastId = blockId; - if (level > TO_BLOCK) lastId++; - while (nextId < lastId) - { - NormBlock* block = block_buffer.Find(nextId); - if (block) + NormBlockId nextId; + if (GetFirstPending(nextId)) + { + NormBlockId lastId; + GetLastPending(lastId); + if ((level < THRU_OBJECT) && (blockId < lastId)) lastId = blockId; + if (level > TO_BLOCK) lastId++; + while (nextId < lastId) { - if ((nextId == blockId) && (THRU_SEGMENT == level)) + if (nextId > lastId) break; + NormBlock* block = block_buffer.Find(nextId); + if (block) { - if (block->FirstPending() <= segmentId) + if ((nextId == blockId) && (THRU_SEGMENT == level)) + { + NormSymbolId firstPending = 0; + if (!block->GetFirstPending(firstPending)) ASSERT(0); + if (firstPending <= segmentId) + { + block->ClearRepairs(); + needRepair = true; + } + } + else { block->ClearRepairs(); needRepair = true; } - } + } else { - block->ClearRepairs(); needRepair = true; } - } - else - { - needRepair = true; - } - nextId++; - nextId = pending_mask.NextSet(nextId); - } + nextId++; + if (!GetNextPending(nextId)) break; + } // end while (nextId < lastId) + } } switch (level) { @@ -674,19 +684,18 @@ bool NormObject::IsRepairPending(bool flush) if (pending_info && !repair_info) return true; // Calculate repair_mask = pending_mask - repair_mask repair_mask.XCopy(pending_mask); - if (repair_mask.IsSet()) + NormBlockId nextId; + if (GetFirstRepair(nextId)) { - NormBlockId nextId = repair_mask.FirstSet(); - NormBlockId lastId = repair_mask.LastSet(); - if (!flush && (current_block_id < lastId)) lastId = current_block_id; - while (nextId <= lastId) + do { + if (!flush && (nextId > current_block_id)) break; NormBlock* block = block_buffer.Find(nextId); if (block) { bool isPending; UINT16 numData = GetBlockSize(nextId); - if (flush || (nextId < lastId)) + if (flush || (nextId < current_block_id)) { isPending = block->IsRepairPending(numData, nparity); } @@ -704,8 +713,7 @@ bool NormObject::IsRepairPending(bool flush) return true; // We need the whole thing } nextId++; - nextId = repair_mask.NextSet(nextId); - } + } while (GetNextRepair(nextId)); } return false; } // end NormObject::IsRepairPending() @@ -716,48 +724,34 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, // If !flush, we request only up _to_ max_pending_block::max_pending_segment. NormRepairRequest req; NormRepairRequest::Form prevForm = NormRepairRequest::INVALID; - NormBlockId prevId; - UINT16 reqCount = 0; - if (pending_mask.IsSet()) + NormBlockId nextId; + bool iterating = GetFirstPending(nextId); + if (iterating) { - NormBlockId nextId = pending_mask.FirstSet(); - NormBlockId lastId = pending_mask.LastSet(); - if (!flush && (max_pending_block < lastId)) lastId = max_pending_block; - lastId++; - DMSG(6, "NormObject::AppendRepairRequest() node>%lu obj>%hu, blk>%lu->%lu (maxPending:%lu)\n", - LocalNodeId(), (UINT16)id, - (UINT32)nextId, (UINT32)lastId, (UINT32)max_pending_block); - - // (TBD) simplify this loop code - while ((nextId <= lastId) || (reqCount > 0)) - { - NormBlock* block = NULL; - bool blockPending = false; - if (nextId == lastId) - nextId++; // force break of possible ending consec. series - else - block = block_buffer.Find(nextId); + iterating = iterating && (flush || (nextId <= max_pending_block)); + NormBlockId prevId = nextId; + UINT32 consecutiveCount = 0; + while (iterating || (0 != consecutiveCount)) + { + NormBlockId lastId; + ASSERT(GetLastPending(lastId)); + DMSG(6, "NormObject::AppendRepairRequest() node>%lu obj>%hu, blk>%lu->%lu (maxPending:%lu)\n", + LocalNodeId(), (UINT16)id, + (UINT32)nextId, (UINT32)lastId, (UINT32)max_pending_block); + bool appendRequest = false; + NormBlock* block = iterating ? block_buffer.Find(nextId) : NULL; if (block) + appendRequest = true; + else if (iterating && ((UINT32)(nextId - prevId) == consecutiveCount)) { - if (nextId == max_pending_block) - { - if (block->FirstPending() < max_pending_segment) - blockPending = true; - } - else - { - blockPending = true; - } - } // end if (block) - - if (!blockPending && reqCount && (reqCount == (nextId - prevId))) - { - reqCount++; // consecutive series of missing blocks continues + consecutiveCount++; } else + appendRequest = true; + if (appendRequest) { NormRepairRequest::Form nextForm; - switch(reqCount) + switch(consecutiveCount) { case 0: nextForm = NormRepairRequest::INVALID; @@ -789,64 +783,89 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, switch (nextForm) { case NormRepairRequest::ITEMS: - req.AppendRepairItem(id, prevId, ndata, 0); // (TBD) error check - if (2 == reqCount) - req.AppendRepairItem(id, prevId+1, ndata, 0); // (TBD) error check + req.AppendRepairItem(id, prevId, GetBlockSize(prevId), 0); // (TBD) error check + if (2 == consecutiveCount) + { + prevId++; + req.AppendRepairItem(id, prevId, GetBlockSize(prevId), 0); // (TBD) error check + } break; case NormRepairRequest::RANGES: - req.AppendRepairItem(id, prevId, ndata, 0); // (TBD) error check - req.AppendRepairItem(id, prevId+reqCount-1, ndata, 0); // (TBD) error check + { + NormBlockId lastId = prevId+consecutiveCount-1; + req.AppendRepairRange(id, prevId, GetBlockSize(prevId), 0, + id, lastId, GetBlockSize(lastId), 0); // (TBD) error check break; + } default: break; } // end switch(nextForm) - prevId = nextId; - if (block || (nextId >= lastId)) - reqCount = 0; - else - reqCount = 1; - } // end if/else (!blockPending && reqCount && (reqCount == (nextId - prevId))) - if (blockPending) - { - UINT16 numData = GetBlockSize(nextId); - if (NormRepairRequest::INVALID != prevForm) - nack.PackRepairRequest(req); // (TBD) error check - reqCount = 0; - prevForm = NormRepairRequest::INVALID; - if (flush || (nextId != max_pending_block)) + if (block) { - block->AppendRepairRequest(nack, numData, nparity, id, - pending_info, segment_size); // (TBD) error check - } - else - { - if (max_pending_segment < numData) - block->AppendRepairRequest(nack, max_pending_segment, 0, id, - pending_info, segment_size); // (TBD) error check + bool blockPending = false; + if (nextId == max_pending_block) + { + NormSymbolId firstPending = 0; + if (!block->GetFirstPending(firstPending)) ASSERT(0); + if (firstPending < max_pending_segment) blockPending = true; + } else - block->AppendRepairRequest(nack, numData, nparity, id, - pending_info, segment_size); // (TBD) error check + { + blockPending = true; + } + if (blockPending) + { + UINT16 numData = GetBlockSize(nextId); + if (NormRepairRequest::INVALID != prevForm) + nack.PackRepairRequest(req); // (TBD) error check + if (flush || (nextId != max_pending_block)) + { + block->AppendRepairRequest(nack, numData, nparity, id, + pending_info, segment_size); // (TBD) error check + } + else + { + if (max_pending_segment < numData) + block->AppendRepairRequest(nack, max_pending_segment, 0, id, + pending_info, segment_size); // (TBD) error check + else + block->AppendRepairRequest(nack, numData, nparity, id, + pending_info, segment_size); // (TBD) error check + } + } + consecutiveCount = 0; + prevForm = NormRepairRequest::INVALID; } - } + else if (iterating) + { + consecutiveCount = 1; + } + else + { + consecutiveCount = 0; // we're all done + } + prevId = nextId; + } // end if (appendRequest) nextId++; - if (nextId <= lastId) - nextId = pending_mask.NextSet(nextId); - } // end while (nextId <= lastId) - if (NormRepairRequest::INVALID != prevForm) - nack.PackRepairRequest(req); // (TBD) error check - } + iterating = GetNextPending(nextId); + //TRACE("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"); + } else { // INFO_ONLY repair request ASSERT(pending_info); nack.AttachRepairRequest(req, segment_size); + prevForm = NormRepairRequest::ITEMS; req.SetForm(NormRepairRequest::ITEMS); req.ResetFlags(); req.SetFlag(NormRepairRequest::INFO); - req.AppendRepairItem(id, 0, ndata, 0); // (TBD) error check - nack.PackRepairRequest(req); // (TBD) error check - } // end if/else pending_mask.IsSet() - + req.AppendRepairItem(id, 0, 0, 0); // (TBD) error check + } // end if/else (iterating) + if (NormRepairRequest::INVALID != prevForm) + nack.PackRepairRequest(req); // (TBD) error check return true; } // end NormObject::AppendRepairRequest() @@ -900,9 +919,9 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, while (!stream->StreamUpdateStatus(blockId)) { // Server is too far ahead of me ... - if (pending_mask.IsSet()) + NormBlockId firstId; + if (GetFirstPending(firstId)) { - NormBlockId firstId = pending_mask.FirstSet(); NormBlock* block = block_buffer.Find(firstId); if (block) { @@ -959,10 +978,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, #ifdef SIMULATE // For simulations, we may need to cap the payloadLength payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); + payloadLength = MIN(payloadLength, SIM_PAYLOAD_MAX); #endif // SIMULATE + memcpy(segment, data.GetPayload(), payloadLength); if (payloadLength < payloadMax) memset(segment+payloadLength, 0, payloadMax-payloadLength); - memcpy(segment, data.GetPayload(), payloadLength); if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) segment[payloadMax] = NormObjectMsg::FLAG_MSG_START; else @@ -987,33 +1007,35 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, // and write any decoded data segments to object DMSG(8, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu blk>%lu " "completed block ...\n", LocalNodeId(), server->GetId(), - (UINT16)id, (UINT32)block->Id()); - UINT16 nextErasure = block->FirstPending(); - UINT16 erasureCount = 0; - if (nextErasure < numData) + (UINT16)id, (UINT32)block->Id()); + UINT16 erasureCount = 0; + UINT16 nextErasure = 0; + if (block->GetFirstPending(nextErasure)) { - UINT16 blockLen = numData + nparity; - while (nextErasure < blockLen) + if (nextErasure < numData) { - server->SetErasureLoc(erasureCount++, nextErasure); - if (nextErasure < numData) + do { - if (!(segment = server->GetFreeSegment(id, blockId))) + server->SetErasureLoc(erasureCount++, nextErasure); + if (nextErasure < numData) { - DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " - "Warning! no free segments ...\n", LocalNodeId(), server->GetId(), - (UINT16)id); - // (TBD) Dump the block ...??? - return; - } - UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); + if (!(segment = server->GetFreeSegment(id, blockId))) + { + DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " + "Warning! no free segments ...\n", LocalNodeId(), server->GetId(), + (UINT16)id); + // (TBD) Dump the block ...??? + return; + } + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); #endif // SIMULATE - memset(segment, 0, payloadMax+1); - block->SetSegment(nextErasure, segment); - } - nextErasure = block->NextPending(nextErasure+1); + memset(segment, 0, payloadMax+1); + block->SetSegment(nextErasure, segment); + } + nextErasure++; + } while (block->GetNextPending(nextErasure)); } } if (erasureCount) @@ -1188,11 +1210,10 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) pending_info = false; return true; } - if (!pending_mask.IsSet()) return false; + NormBlockId blockId; + if (!GetFirstPending(blockId)) return false; NormDataMsg* data = (NormDataMsg*)msg; - - NormBlockId blockId = pending_mask.FirstSet(); UINT16 numData = GetBlockSize(blockId); NormBlock* block = block_buffer.Find(blockId); if (!block) @@ -1239,8 +1260,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) ASSERT(success); } } - ASSERT(block->IsPending()); - NormSegmentId segmentId = block->FirstPending(); + NormSegmentId segmentId = 0; + if (!block->GetFirstPending(segmentId)) ASSERT(0); // Try to read segment if (segmentId < numData) @@ -1264,7 +1285,7 @@ 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); } // Perform incremental FEC encoding as needed @@ -1596,7 +1617,6 @@ UINT16 NormFileObject::ReadSegment(NormBlockId blockId, len = segment_size; } - // Determine segment offset from blockId::segmentId NormObjectSize segmentOffset; NormObjectSize segmentSize = NormObjectSize(segment_size); @@ -1800,7 +1820,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) // Handle potential sync block id wrap NormBlockId delta = stream_next_id - stream_sync_id; if (delta > NormBlockId(2*pending_mask.Size())) - stream_sync_id = NormBlockId(pending_mask.FirstSet()); + GetFirstPending(stream_sync_id); return true; } else @@ -1870,16 +1890,16 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, ASSERT(segment != NULL); UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); ASSERT(segmentLength <= segment_size); -#ifdef SIMULATE - UINT16 simLen = segmentLength + NormDataMsg::GetStreamPayloadHeaderLength(); - simLen = MIN(simLen, SIM_PAYLOAD_MAX); - buffer[simLen] = segment[simLen]; -#else - buffer[segment_size+NormDataMsg::GetStreamPayloadHeaderLength()] = - segment[segment_size+NormDataMsg::GetStreamPayloadHeaderLength()]; -#endif // if/else SIMULATE + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); UINT16 payloadLength = segmentLength+NormDataMsg::GetStreamPayloadHeaderLength(); +#ifdef SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); + UINT16 copyMax = MIN(payloadMax, payloadLength); + memcpy(buffer, segment, copyMax); +#else memcpy(buffer, segment, payloadLength); +#endif // SIMULATE + buffer[payloadMax] = segment[payloadMax]; return payloadLength; } // end NormStreamObject::ReadSegment() @@ -1920,7 +1940,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, { // Notify app for stream data salvage read_index.block = block->Id(); - read_index.segment = block->FirstPending(); + block->GetFirstPending(read_index.segment); NormBlock* tempBlock = block; UINT32 tempOffset = read_offset; session->Notify(NormController::RX_OBJECT_UPDATE, server, this); @@ -1974,14 +1994,14 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, { char* s = segment_pool.Get(); ASSERT(s != NULL); // for now, this should always succeed - UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); - UINT16 flagsOffset = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); + UINT16 payloadLength = NormDataMsg::ReadStreamPayloadLength(segment); + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - flagsOffset = MIN(SIM_PAYLOAD_MAX, flagsOffset); - segmentLength = MIN((SIM_PAYLOAD_MAX-NormDataMsg::GetStreamPayloadHeaderLength()), segmentLength); + payloadMax = MIN(SIM_PAYLOAD_MAX, payloadMax); + payloadLength = MIN(payloadMax, payloadLength); #endif // SIMULATE - s[flagsOffset] = segment[flagsOffset]; - memcpy(s, segment, segmentLength + NormDataMsg::GetStreamPayloadHeaderLength()); + memcpy(s, segment, payloadLength); + s[payloadMax] = segment[payloadMax]; block->AttachSegment(segmentId, s); block->SetPending(segmentId); ASSERT(block->Segment(segmentId) == s); @@ -2007,9 +2027,9 @@ void NormStreamObject::Prune(NormBlockId blockId) break; } } - if (pending_mask.IsSet()) + NormBlockId firstId; + if (GetFirstPending(firstId)) { - NormBlockId firstId = pending_mask.FirstSet(); if (firstId < blockId) { resync = true; @@ -2079,11 +2099,11 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar } else { - UINT16 flagsOffset = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); + UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - flagsOffset = MIN(flagsOffset, SIM_PAYLOAD_MAX); + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); #endif // if/else SIMULATE - msgStart = (NormObjectMsg::FLAG_MSG_START == segment[flagsOffset]); + msgStart = (NormObjectMsg::FLAG_MSG_START == segment[payloadMax]); } if (!msgStart) { @@ -2103,9 +2123,8 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar } #ifdef SIMULATE - UINT16 simCount = ((index+count) < SIM_PAYLOAD_MAX) ? - count : ((index < SIM_PAYLOAD_MAX) ? - (SIM_PAYLOAD_MAX - index) : 0); + UINT16 simCount = index + count + NormDataMsg::GetStreamPayloadHeaderLength(); + simCount = (simCount < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - simCount) : 0; memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), simCount); #else memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), count); @@ -2187,7 +2206,7 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, if (c) { block_buffer.Remove(c); - session->ServerPutFreeBlock(c); + session->ServerPutFreeBlock(c); } } else if (b->IsPending()) @@ -2217,11 +2236,11 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, msgStartValue = NormObjectMsg::FLAG_MSG_START; msg_start = false; } - UINT16 flagsOffset = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); + UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE - flagsOffset = MIN(SIM_PAYLOAD_MAX, flagsOffset); -#endif // SIMULATE - segment[flagsOffset] = msgStartValue; + payloadMax = MIN(SIM_PAYLOAD_MAX, payloadMax); +#endif // SIMULATE + segment[payloadMax] = msgStartValue; } else { @@ -2232,11 +2251,13 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, UINT16 space = segment_size - index; count = MIN(count, space); #ifdef SIMULATE - UINT16 simCount = ((index+count) < SIM_PAYLOAD_MAX) ? - count : ((index < SIM_PAYLOAD_MAX) ? - (SIM_PAYLOAD_MAX - index) : 0); + UINT16 simCount = index + NormDataMsg::GetStreamPayloadHeaderLength(); + simCount = (simCount < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - simCount) : 0; + 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); @@ -2275,7 +2296,7 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len, #ifdef SIMULATE ///////////////////////////////////////////////////////////////// // -// NormSimObject Implementation +// NormSimObject Implementation (dummy NORM_OBJECT_FILE or NORM_OBJECT_DATA) // NormSimObject::NormSimObject(class NormSession* theSession, diff --git a/common/normObject.h b/common/normObject.h index 9006815..8abc995 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -70,7 +70,50 @@ class NormObject bool IsPending(bool flush = true) const; bool IsRepairPending() const; bool IsPendingInfo() {return pending_info;} - NormBlockId FirstPending() {return pending_mask.FirstSet();} + bool GetFirstPending(NormBlockId& blockId) const + { + unsigned long index = 0; + bool result = pending_mask.GetFirstSet(index); + blockId = NormBlockId(index); + return result; + } + bool GetNextPending(NormBlockId& blockId) const + { + unsigned long index = (UINT32)blockId; + bool result = pending_mask.GetNextSet(index); + blockId = NormBlockId(index); + return result; + } + bool GetLastPending(NormBlockId& blockId) const + { + unsigned long index = 0; + bool result = pending_mask.GetLastSet(index); + blockId = NormBlockId(index); + return result; + } + bool GetFirstRepair(NormBlockId& blockId) const + { + unsigned long index = 0; + bool result = repair_mask.GetFirstSet(index); + blockId = NormBlockId(index); + return result; + } + + bool GetNextRepair(NormBlockId& blockId) const + { + unsigned long index = (UINT32)blockId; + bool result = repair_mask.GetNextSet(index); + blockId = NormBlockId(index); + return result; + } + bool GetLastRepair(NormBlockId& blockId) const + { + unsigned long index = 0; + bool result = repair_mask.GetLastSet(index); + blockId = NormBlockId(index); + return result; + } + // Methods available to server for transmission bool NextServerMsg(NormObjectMsg* msg); diff --git a/common/normSegment.cpp b/common/normSegment.cpp index c16fa49..5990a07 100644 --- a/common/normSegment.cpp +++ b/common/normSegment.cpp @@ -18,12 +18,11 @@ bool NormSegmentPool::Init(unsigned int count, unsigned int size) { if (seg_list) Destroy(); peak_usage = 0; - overruns = 0; - + overruns = 0; #ifdef SIMULATE // In simulations, don't really need big vectors for data - // since we don't actually read/write real data - size = MIN(size, (SIM_PAYLOAD_MAX+NormDataMsg::PayloadHeaderLength()+1)); + // since we don't actually read/write real data (for the most part) + size = MIN(size, (SIM_PAYLOAD_MAX+1)); #endif // SIMULATE // This makes sure we get appropriate alignment unsigned int alloc_size = size / sizeof(char*); @@ -184,14 +183,15 @@ bool NormBlock::IsRepairPending(UINT16 ndata, UINT16 nparity) if (nparity) { UINT16 i = nparity; - NormSegmentId nextId = (UINT16)pending_mask.FirstSet(); + NormSegmentId nextId = 0; + GetFirstPending(nextId); while (i--) { // (TBD) for more NACK suppression, we could skip ahead // if this bit is already set in repair_mask? repair_mask.Set(nextId); // set bit a parity can fill nextId++; - nextId = (UINT16)pending_mask.NextSet(nextId); + GetNextPending(nextId); } } else if (size > ndata) @@ -390,66 +390,68 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd, NormRepairRequest req; req.SetFlag(NormRepairRequest::SEGMENT); if (repairInfo) req.SetFlag(NormRepairRequest::INFO); - UINT16 nextId = (UINT16)repair_mask.FirstSet(); - UINT16 blockSize = size; - NormRepairRequest::Form prevForm = NormRepairRequest::INVALID; - UINT16 segmentCount = 0; - UINT16 firstId = 0; - while (nextId < blockSize) + NormSymbolId nextId = 0; + if (GetFirstRepair(nextId)) { - UINT16 currentId = nextId; - nextId = (UINT16)repair_mask.NextSet(++nextId); - if (!segmentCount) firstId = currentId; - - segmentCount++; - // Check for break in consecutive series or end - if (((nextId - currentId) > 1) || (nextId >= blockSize)) + UINT16 blockSize = size; + NormRepairRequest::Form prevForm = NormRepairRequest::INVALID; + UINT16 segmentCount = 0; + UINT16 firstId = 0; + while (nextId < blockSize) { - NormRepairRequest::Form form; - switch (segmentCount) + UINT16 currentId = nextId; + if (!GetNextRepair(++nextId)) nextId = blockSize; + if (!segmentCount) firstId = currentId; + segmentCount++; + // Check for break in consecutive series or end + if (((nextId - currentId) > 1) || (nextId >= blockSize)) { - case 0: - form = NormRepairRequest::INVALID; - break; - case 1: - case 2: - form = NormRepairRequest::ITEMS; - break; - default: - form = NormRepairRequest::RANGES; - break; + NormRepairRequest::Form form; + switch (segmentCount) + { + case 0: + form = NormRepairRequest::INVALID; + break; + case 1: + case 2: + form = NormRepairRequest::ITEMS; + break; + default: + form = NormRepairRequest::RANGES; + break; + } + if (form != prevForm) + { + if (NormRepairRequest::INVALID != prevForm) + cmd.PackRepairRequest(req); // (TBD) error check + req.SetForm(form); + cmd.AttachRepairRequest(req, segmentSize); // (TBD) error check + prevForm = form; + } + switch(form) + { + case NormRepairRequest::INVALID: + ASSERT(0); // can't happen + break; + case NormRepairRequest::ITEMS: + req.AppendRepairItem(objectId, id, ndata, firstId); + if (2 == segmentCount) + req.AppendRepairItem(objectId, id, ndata, currentId); + break; + case NormRepairRequest::RANGES: + req.AppendRepairRange(objectId, id, ndata, firstId, + objectId, id, ndata, currentId); + break; + case NormRepairRequest::ERASURES: + // erasure counts not used + break; + } + segmentCount = 0; } - if (form != prevForm) - { - if (NormRepairRequest::INVALID != prevForm) - cmd.PackRepairRequest(req); // (TBD) error check - req.SetForm(form); - cmd.AttachRepairRequest(req, segmentSize); // (TBD) error check - prevForm = form; - } - switch(form) - { - case NormRepairRequest::INVALID: - ASSERT(0); // can't happen - break; - case NormRepairRequest::ITEMS: - req.AppendRepairItem(objectId, id, ndata, firstId); - if (2 == segmentCount) - req.AppendRepairItem(objectId, id, ndata, currentId); - break; - case NormRepairRequest::RANGES: - req.AppendRepairRange(objectId, id, ndata, firstId, - objectId, id, ndata, currentId); - break; - case NormRepairRequest::ERASURES: - // erasure counts not used - break; - } - segmentCount = 0; - } - } // end while (nextId < blockSize) - if (NormRepairRequest::INVALID != prevForm) - cmd.PackRepairRequest(req); // (TBD) error check + } // end while (nextId < blockSize) + if (NormRepairRequest::INVALID != prevForm) + cmd.PackRepairRequest(req); // (TBD) error check + } return true; } // end NormBlock::AppendRepairAdv() @@ -461,25 +463,25 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, bool pendingInfo, UINT16 segmentSize) { - ASSERT(pending_mask.FirstSet() < ndata); - NormSegmentId nextId, endId; - + NormSegmentId nextId = 0; + NormSegmentId endId; if (erasure_count > nparity) { // Request explicit repair - nextId = (UINT16)pending_mask.FirstSet(); + GetFirstPending(nextId); UINT16 i = nparity; // Skip nparity missing data segments while (i--) { nextId++; - endId = (UINT16)pending_mask.NextSet(nextId); + GetNextPending(nextId); } endId = ndata + nparity; } else { - nextId = (UINT16)pending_mask.NextSet(ndata); + nextId = ndata; + GetNextPending(nextId); endId = ndata + erasure_count; } NormRepairRequest req; @@ -492,7 +494,7 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, while (nextId < endId) { UINT16 currentId = nextId; - nextId = (UINT16)pending_mask.NextSet(++nextId); + if (!GetNextPending(++nextId)) nextId = endId; if (0 == segmentCount) firstId = currentId; segmentCount++; // Check for break in consecutive series or end @@ -796,7 +798,9 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock) prev->next = theBlock; else table[index] = theBlock; + ASSERT((entry ? (blockId != entry->Id()) : true)); + theBlock->next = entry; return true; } // end NormBlockBuffer::Insert() diff --git a/common/normSegment.h b/common/normSegment.h index 158dbc0..c824458 100644 --- a/common/normSegment.h +++ b/common/normSegment.h @@ -156,10 +156,36 @@ class NormBlock void IncrementParityCount() {parity_count++;} UINT16 ParityCount() const {return parity_count;} - NormSymbolId FirstPending() const - {return (NormSymbolId)pending_mask.FirstSet();} - NormSymbolId FirstRepair() const - {return (NormSymbolId)repair_mask.FirstSet();} + bool GetFirstPending(NormSymbolId& symbolId) const + { + UINT32 index; + bool result = pending_mask.GetFirstSet(index); + symbolId = (UINT16)index; + return result; + } + bool GetNextPending(NormSymbolId& symbolId) const + { + UINT32 index = (UINT32)symbolId; + bool result = pending_mask.GetNextSet(index); + symbolId = (UINT16)index; + return result; + } + NormSymbolId GetFirstRepair(NormSymbolId& symbolId) const + { + UINT32 index; + bool result = repair_mask.GetFirstSet(index); + symbolId = (UINT16)index; + return result; + } + + bool GetNextRepair(NormSymbolId& symbolId) const + { + UINT32 index = (UINT32)symbolId; + bool result = repair_mask.GetNextSet(index); + symbolId = (UINT16)index; + return result; + } + bool SetPending(NormSymbolId s) {return pending_mask.Set(s);} bool SetPending(NormSymbolId firstId, UINT16 count) @@ -190,11 +216,6 @@ class NormBlock bool IsTransmitPending() const {return (pending_mask.IsSet() || repair_mask.IsSet());} - - NormSymbolId NextPending(NormSymbolId index) const - {return ((NormSymbolId)pending_mask.NextSet(index));} - - bool AppendRepairRequest(NormNackMsg& nack, UINT16 ndata, UINT16 nparity, diff --git a/common/normSession.cpp b/common/normSession.cpp index 2cb520f..45c2bf2 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -170,13 +170,13 @@ bool NormSession::StartServer(unsigned long bufferSpace, StopServer(); return false; } - if (!tx_pending_mask.Init(256)) + if (!tx_pending_mask.Init(256, 0x0000ffff)) { DMSG(0, "NormSession::StartServer() tx_pending_mask.Init() error!\n"); StopServer(); return false; } - if (!tx_repair_mask.Init(256)) + if (!tx_repair_mask.Init(256, 0x0000ffff)) { DMSG(0, "NormSession::StartServer() tx_repair_mask.Init() error!\n"); StopServer(); @@ -278,9 +278,9 @@ void NormSession::Serve() NormObject* obj = NULL; // Queue next server message - if (tx_pending_mask.IsSet()) + NormObjectId objectId; + if (ServerGetFirstPending(objectId)) { - NormObjectId objectId((unsigned short)tx_pending_mask.FirstSet()); obj = tx_table.Find(objectId); ASSERT(obj); } @@ -937,12 +937,9 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast) if (IsClient()) ClientHandleNackMessage((NormNackMsg&)msg); break; case NormMsg::ACK: - DMSG(4, "NormSession::HandleReceiveMessage(NormMsg::ACK) node>%lu ...\n", - LocalNodeId()); if (IsServer() && (((NormAckMsg&)msg).GetServerId() == LocalNodeId())) ServerHandleAckMessage(currentTime, (NormAckMsg&)msg, wasUnicast); - if (IsClient()) - ClientHandleAckMessage((NormAckMsg&)msg); + if (IsClient()) ClientHandleAckMessage((NormAckMsg&)msg); break; case NormMsg::REPORT: @@ -1048,7 +1045,6 @@ double NormSession::CalculateRtt(const struct timeval& currentTime, void NormSession::ServerUpdateGrttEstimate(double clientRtt) { - grtt_response = true; if (clientRtt > grtt_current_peak) { @@ -1361,23 +1357,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor NormObjectId txObjectIndex; NormBlockId txBlockIndex; - if (tx_pending_mask.IsSet()) + if (ServerGetFirstPending(txObjectIndex)) { - txObjectIndex = NormObjectId((unsigned short)tx_pending_mask.FirstSet()); NormObject* obj = tx_table.Find(txObjectIndex); - ASSERT(obj); - if (obj->IsPending()) + ASSERT(obj && obj->IsPending()); + if (obj->IsPendingInfo()) { - if (obj->IsPendingInfo()) - txBlockIndex = 0; - else - txBlockIndex = obj->FirstPending() + 1; + txBlockIndex = 0; } else { - txObjectIndex = txObjectIndex + 1; - txBlockIndex = 0; - } + obj->GetFirstPending(txBlockIndex); + txBlockIndex++; + } } else { @@ -1447,8 +1439,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor ServerQueueSquelch(nextObjectId); squelchQueued = true; } - if ((OBJECT == requestLevel) || - (INFO == requestLevel)) + if ((OBJECT == requestLevel) || (INFO == requestLevel)) { nextObjectId++; if (nextObjectId > lastObjectId) inRange = false; @@ -1481,6 +1472,8 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor switch (requestLevel) { case OBJECT: + DMSG(8, "NormSession::ServerHandleNackMessage(OBJECT) objs>%hu:%hu\n", + (UINT16)nextObjectId, (UINT16)lastObjectId); if (holdoff) { if (nextObjectId > txObjectIndex) @@ -1504,6 +1497,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor break; case BLOCK: + DMSG(8, "NormSession::ServerHandleNackMessage(BLOCK) obj>%hu blks>%lu:%lu\n", + (UINT16)nextObjectId, (UINT32)nextBlockId, (UINT32)lastBlockId); + // (TBD) if entire object is TxReset(), continue if (object->IsStream()) { @@ -1577,6 +1573,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor inRange = false; break; case SEGMENT: + DMSG(8, "NormSession::ServerHandleNackMessage(SEGMENT) obj>%hu blk>%lu segs>%hu:%hu\n", + (UINT16)nextObjectId, (UINT32)nextBlockId, + (UINT32)nextSegmentId, (UINT32)lastSegmentId); if (nextBlockId != prevBlockId) freshBlock = true; if (freshBlock) { @@ -1635,7 +1634,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor if (1 == (txBlockIndex - nextBlockId)) { // We're currently sending this block - NormSegmentId firstPending = block->FirstPending(); + ASSERT(block->IsPending()); + NormSegmentId firstPending = 0; + block->GetFirstPending(firstPending); if (lastLockId <= firstPending) attemptLock = false; else if (nextSegmentId < firstPending) @@ -1696,7 +1697,8 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } else if (1 == (txBlockIndex - nextBlockId)) { - NormSegmentId firstPending = block->FirstPending(); + NormSegmentId firstPending = 0; + if (!block->GetFirstPending(firstPending)) ASSERT(0); if (nextSegmentId > firstPending) object->TxUpdateBlock(block, nextSegmentId, lastSegmentId, numErasures); else if (lastSegmentId > firstPending) @@ -2223,8 +2225,12 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) } // 2) Update grtt_estimate _if_ sufficient time elapsed. + // This new code allows more liberal downward adjustment of + // of grtt when congestion control is enabled. grtt_age += probe_timer.GetInterval(); - if (grtt_age >= grtt_interval) + double ageMax = 3 * grtt_advertised; + ageMax = ageMax > grtt_interval_min ? ageMax : grtt_interval_min; + if (grtt_age >= ageMax)//grtt_interval) { if (grtt_response) { @@ -2266,7 +2272,7 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/) if (grtt_interval < grtt_interval_min) grtt_interval = grtt_interval_min; else - grtt_interval *= 2.0; + grtt_interval *= 1.5; if (grtt_interval > grtt_interval_max) grtt_interval = grtt_interval_max; @@ -2417,13 +2423,19 @@ bool NormSession::OnReportTimeout(ProtoTimer& /*theTimer*/) struct tm* ct = gmtime((time_t*)¤tTime.tv_sec); DMSG(2, "Report time>%02d:%02d:%02d.%06lu node>%lu ***************************************\n", ct->tm_hour, ct->tm_min, ct->tm_sec, currentTime.tv_usec, LocalNodeId()); + if (IsServer()) + { + DMSG(2, "Local server:\n"); + DMSG(2, " tx_rate>%9.3lf kbps grtt>%lf\n", + ((double)tx_rate)*8.0/1000.0, grtt_advertised); + } if (IsClient()) { NormNodeTreeIterator iterator(server_tree); NormServerNode* next; while ((next = (NormServerNode*)iterator.GetNextNode())) { - DMSG(2, "Remote server:%lu\n", next->GetId()); + DMSG(2, "Remote server>%lu\n", next->GetId()); double rxRate = 8.0e-03*((double)next->RecvTotal()) / report_timer.GetInterval(); // kbps double rxGoodput = 8.0e-03*((double)next->RecvGoodput()) / report_timer.GetInterval(); // kbps next->ResetRecvStats(); diff --git a/common/normSession.h b/common/normSession.h index 1f527bb..89074a3 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -154,6 +154,14 @@ class NormSession void ServerSetExtraParity(UINT16 extraParity) {extra_parity = extraParity;} + bool ServerGetFirstPending(NormObjectId& objectId) + { + UINT32 index; + bool result = tx_pending_mask.GetFirstSet(index); + objectId = (UINT16)index; + return result; + } + double ServerGroupSize() {return gsize_measured;} void ServerSetGroupSize(double gsize) { diff --git a/common/normVersion.h b/common/normVersion.h index ed93263..34c9b04 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.1b2" +#define VERSION "1.1b3" #endif // _NORM_VERSION diff --git a/ns/simplenorm.tcl b/ns/simplenorm.tcl index 7482bc1..fc3c7e9 100644 --- a/ns/simplenorm.tcl +++ b/ns/simplenorm.tcl @@ -121,7 +121,7 @@ $norm(1) repeat 50 $norm(1) interval 0.0 $norm(1) txloss 10.0 $norm(1) gsize $groupSize -$norm(1) cc off +$norm(1) cc on #$norm(1) trace on # 8) Configure NORM client agents at other nodes @@ -141,7 +141,7 @@ for {set i 2} {$i <= $numNodes} {incr i} { $ns_ at 0.0 "$norm($i) start client" } -$ns_ at 0.0 "$norm(1) sendFile 64000" +$ns_ at 0.0 "$norm(1) sendFile 1000000" $ns_ at $duration "finish $ns_ $f $nf $nsTracing"