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

v1.1b3

pull/2/head
Jeff Weston 2019-09-11 11:31:13 -04:00
parent 541a4bfa9b
commit c74826c08f
18 changed files with 800 additions and 622 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@ -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- a set of files _or_ a stream piped to/from stdin/stdout. The command-
line options are not yet fully documented. line options are not yet fully documented.
I do have a PDF file which explains some of the command-line options. The current code only has Makefiles for Unix platforms, but the
Email me for it. code could be assembled as Win32 project under VC++.
We're getting there ;-)
The NORM code depends upon the current "Protolib" release. The NORM code depends upon the current "Protolib" release.
See <http://pf.itd.nrl.navy.mil> for that code. See <http://pf.itd.nrl.navy.mil> for that code.

7
VERSION.TXT Normal file
View File

@ -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.

7
common/normApp.cpp
View File

@ -966,7 +966,10 @@ void NormApp::Notify(NormController::Event event,
else else
{ {
if (readLength > 0) 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) if (readLength)
@ -1110,7 +1113,7 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/)
if (tx_repeat_interval > tx_object_interval) if (tx_repeat_interval > tx_object_interval)
{ {
if (interval_timer.IsActive()) interval_timer.Deactivate(); 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); ActivateTimer(interval_timer);
return false; return false;
} }

300
common/normBitmask.cpp
View File

@ -199,10 +199,10 @@ void NormBitmask::Destroy()
} // end 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 >= num_bits) return false;
if (index < first_set) return first_set; if (index < first_set) return GetFirstSet(index);
unsigned long maskIndex = index >> 3; unsigned long maskIndex = index >> 3;
if (mask[maskIndex]) if (mask[maskIndex])
{ {
@ -211,22 +211,28 @@ unsigned long NormBitmask::NextSet(unsigned long index) const
for (int i = 0; i < w; i++) for (int i = 0; i < w; i++)
{ {
int loc = BITLOCS[mask[maskIndex]][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) while(++maskIndex < mask_len)
{ {
if (mask[maskIndex]) 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() } // 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 >= num_bits) index = num_bits - 1;
if (index < first_set) return num_bits; if (index < first_set) return false;
unsigned long maskIndex = index >> 3; unsigned long maskIndex = index >> 3;
if (mask[maskIndex]) if (mask[maskIndex])
{ {
@ -237,7 +243,8 @@ unsigned long NormBitmask::PrevSet(unsigned long index) const
int loc = BITLOCS[mask[maskIndex]][i]; int loc = BITLOCS[mask[maskIndex]][i];
if (loc <= remainder) 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]) if (mask[maskIndex])
{ {
int w = WEIGHT[mask[maskIndex]] - 1; 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() } // 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; if (index >= num_bits) return false;
unsigned long maskIndex = index >> 3; unsigned long next = index;
unsigned char bit = 0x80 >> (index & 0x07); unsigned long maskIndex = next >> 3;
while (index < num_bits) unsigned char bit = 0x80 >> (next & 0x07);
while (next < num_bits)
{ {
unsigned char val = mask[maskIndex]; unsigned char val = mask[maskIndex];
if (val) if (val)
{ {
while (bit && (index < num_bits)) while (bit && (next < num_bits))
{ {
if (0 == (val & bit)) return index; if (0 == (val & bit))
index++; {
index = next;
return true;
}
next++;
bit >>= 0x01; bit >>= 0x01;
} }
bit = 0x80; bit = 0x80;
@ -275,10 +288,11 @@ unsigned long NormBitmask::NextUnset(unsigned long index) const
} }
else else
{ {
return index; index = next;
return true;
} }
} }
return index; return false;
} // end NormBitmask::NextUnset() } // end NormBitmask::NextUnset()
@ -337,8 +351,11 @@ bool NormBitmask::UnsetBits(unsigned long index, unsigned long count)
count &= 0x07; count &= 0x07;
if (count) mask[maskIndex+nbytes] &= 0xff >> count; if (count) mask[maskIndex+nbytes] &= 0xff >> count;
} }
if ((first_set >= index) && (end > first_set)) if ((first_set >= index) && (end > first_set))
first_set = NextSet(end); {
first_set = end;
if (!GetNextSet(first_set)) first_set = num_bits;
}
return true; return true;
} // end NormBitmask::UnsetBits() } // end NormBitmask::UnsetBits()
@ -370,7 +387,9 @@ bool NormBitmask::Subtract(const NormBitmask& b)
for(unsigned long i = 0; i < len; i++) for(unsigned long i = 0; i < len; i++)
mask[i] &= ~b.mask[i]; mask[i] &= ~b.mask[i];
if (first_set >= b.first_set) if (first_set >= b.first_set)
first_set = NextSet(first_set); {
if (!GetNextSet(first_set)) first_set = num_bits;
}
return true; return true;
} // end NormBitmask::Subtract() } // end NormBitmask::Subtract()
@ -386,9 +405,14 @@ bool NormBitmask::XCopy(const NormBitmask& b)
mask[i] = b.mask[i] & ~mask[i]; mask[i] = b.mask[i] & ~mask[i];
if (len < mask_len) memset(&mask[len], 0, mask_len - len); if (len < mask_len) memset(&mask[len], 0, mask_len - len);
if (b.first_set < first_set) if (b.first_set < first_set)
{
first_set = b.first_set; first_set = b.first_set;
}
else else
first_set = NextSet(b.first_set); {
first_set = b.first_set;
GetNextSet(first_set);
}
return true; return true;
} // end NormBitmask::XCopy() } // end NormBitmask::XCopy()
@ -400,9 +424,14 @@ bool NormBitmask::Multiply(const NormBitmask& b)
mask[i] |= b.mask[i]; mask[i] |= b.mask[i];
if (len < mask_len) memset(&mask[len], 0, mask_len - len); if (len < mask_len) memset(&mask[len], 0, mask_len - len);
if (b.first_set > first_set) 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) else if (first_set > b.first_set)
first_set = NextSet(first_set); {
if (!GetNextSet(first_set)) first_set = num_bits;
}
return true; return true;
} // end NormBitmask::Multiply() } // end NormBitmask::Multiply()
@ -413,7 +442,7 @@ bool NormBitmask::Xor(const NormBitmask& b)
for(unsigned int i = 0; i < b.mask_len; i++) for(unsigned int i = 0; i < b.mask_len; i++)
mask[i] ^= b.mask[i]; mask[i] ^= b.mask[i];
if (b.first_set == first_set) if (b.first_set == first_set)
first_set = NextSet(first_set); GetNextSet(first_set);
return true; return true;
} // end NormBitmask::Xor() } // end NormBitmask::Xor()
@ -445,7 +474,7 @@ NormSlidingMask::~NormSlidingMask()
Destroy(); Destroy();
} }
bool NormSlidingMask::Init(long numBits) bool NormSlidingMask::Init(long numBits, UINT32 rangeMask)
{ {
if (mask) Destroy(); if (mask) Destroy();
@ -453,6 +482,8 @@ bool NormSlidingMask::Init(long numBits)
unsigned long len = (numBits + 7) >> 3; unsigned long len = (numBits + 7) >> 3;
if ((mask = new unsigned char[len])) if ((mask = new unsigned char[len]))
{ {
range_mask = rangeMask;
range_sign = (rangeMask ^ (rangeMask >> 1));
mask_len = len; mask_len = len;
num_bits = numBits; num_bits = numBits;
Clear(); Clear();
@ -479,8 +510,8 @@ bool NormSlidingMask::CanSet(unsigned long index) const
if (IsSet()) if (IsSet())
{ {
// Determine position with respect to current start // Determine position with respect to current start
// and end, given the "offset" of the current start // and end, given the "offset" of the current start
long pos = index - offset; long pos = Delta(index, offset);
if (pos < 0) if (pos < 0)
{ {
// Precedes start. // Precedes start.
@ -505,11 +536,12 @@ bool NormSlidingMask::CanSet(unsigned long index) const
return true; return true;
} }
} }
else if (pos < num_bits)
{
return true;
}
else else
{ {
if (pos < num_bits)
return true;
else
return false; // out of range return false; // out of range
} }
} }
@ -526,12 +558,8 @@ bool NormSlidingMask::Set(unsigned long index)
{ {
// Determine position with respect to current start // Determine position with respect to current start
// and end, given the "offset" of the current start // and end, given the "offset" of the current start
long pos = index - offset; long pos = Delta(index , offset);
if (pos <= -num_bits) if (pos < 0)
{
return false; // out-of-range
}
else if (pos < 0)
{ {
// Precedes start. // Precedes start.
pos += start; pos += start;
@ -557,11 +585,7 @@ bool NormSlidingMask::Set(unsigned long index)
offset = index; offset = index;
} }
} }
else if (pos >= num_bits) else if (pos < num_bits)
{
return false; // out of range
}
else
{ {
pos += start; pos += start;
if (pos >= num_bits) pos -= num_bits; if (pos >= num_bits) pos -= num_bits;
@ -574,6 +598,10 @@ bool NormSlidingMask::Set(unsigned long index)
if ((pos > end) || (pos < start)) end = pos; if ((pos > end) || (pos < start)) end = pos;
} }
} }
else
{
return false; // out of range
}
ASSERT((pos >> 3) >= 0); ASSERT((pos >> 3) >= 0);
ASSERT((pos >> 3) < (long)mask_len); ASSERT((pos >> 3) < (long)mask_len);
mask[(pos >> 3)] |= (0x80 >> (pos & 0x07)); mask[(pos >> 3)] |= (0x80 >> (pos & 0x07));
@ -592,11 +620,14 @@ bool NormSlidingMask::Unset(unsigned long index)
//ASSERT(CanSet(index)); //ASSERT(CanSet(index));
if (IsSet()) if (IsSet())
{ {
long pos = index - offset; long pos = Delta(index, offset);
if (pos >= 0) if (pos < 0)
{ {
// Is it in range? return true; // out-of-range
if (pos >= num_bits) return true; }
else if (pos < num_bits)
{
// Is it in current range of set bits?
pos += start; pos += start;
if (pos >= num_bits) pos -= num_bits; if (pos >= num_bits) pos -= num_bits;
if (end < start) if (end < start)
@ -607,6 +638,7 @@ bool NormSlidingMask::Unset(unsigned long index)
{ {
if ((pos < start) || (pos > end)) return true; if ((pos < start) || (pos > end)) return true;
} }
// Yes, it was in range.
// Unset the corresponding bit // Unset the corresponding bit
ASSERT((pos >> 3) >= 0); ASSERT((pos >> 3) >= 0);
ASSERT((pos >> 3) < (long)mask_len); ASSERT((pos >> 3) < (long)mask_len);
@ -619,29 +651,28 @@ bool NormSlidingMask::Unset(unsigned long index)
} }
if (start == pos) if (start == pos)
{ {
unsigned long next = NextSet(index); unsigned long next = index;
long delta = next - offset + start; if (!GetNextSet(next)) ASSERT(0);
if (delta >= num_bits) long delta = Delta(next, offset);
start = delta - num_bits; ASSERT(delta >= 0);
else if (start < 0) start += delta;
start = delta + num_bits; if (start >= num_bits) start -= num_bits;
else
start = delta;
offset = next; offset = next;
} }
if (pos == end) if (pos == end)
{ {
unsigned long prev = PrevSet(index); unsigned long prev = index;
long delta = prev - offset + start; if (!GetPrevSet(prev)) ASSERT(0);
if (delta >= num_bits) long delta = Delta(prev, offset);
end = delta - num_bits; ASSERT(delta >= 0);
else if (delta < 0) end = start + delta;
end = delta + num_bits; if (end >= num_bits) end -= num_bits;
else
end = delta;
} }
} }
else
{
return true; // out-of-range
}
} }
return true; return true;
} // end NormSlidingMask::Unset() } // end NormSlidingMask::Unset()
@ -658,8 +689,8 @@ bool NormSlidingMask::SetBits(unsigned long index, long count)
long last = index + count - 1; long last = index + count - 1;
if (!CanSet(index)) return false; if (!CanSet(index)) return false;
if (!CanSet(last)) return false; if (!CanSet(last)) return false;
// Calculate first set bit position // Calculate first set bit position
firstPos = index - offset; firstPos = Delta(index, offset);
if (firstPos < 0) if (firstPos < 0)
{ {
// precedes start // precedes start
@ -674,7 +705,7 @@ bool NormSlidingMask::SetBits(unsigned long index, long count)
if (firstPos >= num_bits) firstPos -= num_bits; if (firstPos >= num_bits) firstPos -= num_bits;
} }
// Calculate last set bit position // Calculate last set bit position
lastPos = last - offset; lastPos = Delta(last , offset);
if (lastPos > 0) if (lastPos > 0)
{ {
// Is post start, post end? // Is post start, post end?
@ -775,7 +806,7 @@ bool NormSlidingMask::UnsetBits(unsigned long index, long count)
long firstPos; long firstPos;
if (count <= 0) return true; if (count <= 0) return true;
if (count > num_bits) count = num_bits; if (count > num_bits) count = num_bits;
long delta = index - offset; long delta = Delta(index , offset);
if (delta >= num_bits) if (delta >= num_bits)
{ {
return true; return true;
@ -791,7 +822,9 @@ bool NormSlidingMask::UnsetBits(unsigned long index, long count)
firstPos = start + delta; firstPos = start + delta;
if (firstPos >= num_bits) firstPos -= num_bits; 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; long lastPos;
if (delta < 0) if (delta < 0)
{ {
@ -863,17 +896,23 @@ bool NormSlidingMask::UnsetBits(unsigned long index, long count)
mask[maskIndex+nbytes] &= 0xff >> 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 (start == firstPos)
{ {
if (end == lastPos) if (end == lastPos)
{
start = end = num_bits; start = end = num_bits;
}
else else
Unset(FirstSet()); {
Unset(offset);
}
} }
else if (end == lastPos) else if (end == lastPos)
{ {
Unset(LastSet()); UINT32 lastSet;
GetLastSet(lastSet);
Unset(lastSet);
} }
} }
return true; return true;
@ -883,7 +922,7 @@ bool NormSlidingMask::Test(unsigned long index) const
{ {
if (IsSet()) if (IsSet())
{ {
long pos = index - offset; long pos = Delta(index , offset);
if (pos >= 0) if (pos >= 0)
{ {
// Is it in range? // Is it in range?
@ -907,25 +946,26 @@ bool NormSlidingMask::Test(unsigned long index) const
} // end NormSlidingMask::Test() } // end NormSlidingMask::Test()
unsigned long NormSlidingMask::NextSet(unsigned long index) const bool NormSlidingMask::GetNextSet(unsigned long& index) const
{ {
if (IsSet()) if (IsSet())
{ {
long pos = index - offset; unsigned long next = index;
long pos = Delta(next, offset);
if (pos >= 0) if (pos >= 0)
{ {
// Is it in range? // Is it in range?
if (pos >= num_bits) return (offset + num_bits); if (pos >= num_bits) return false;
pos += start; pos += start;
if (pos >= num_bits) pos -= num_bits; if (pos >= num_bits) pos -= num_bits;
if (end < start) if (end < start)
{ {
if ((pos > end) && (pos < start)) return (offset + num_bits); if ((pos > end) && (pos < start)) return false;
} }
else else
{ {
if ((pos < start) || (pos > end)) return (offset + num_bits); if ((pos < start) || (pos > end)) return false;
} }
// Seek next set bit // Seek next set bit
unsigned long maskIndex = pos >> 3; unsigned long maskIndex = pos >> 3;
@ -941,7 +981,8 @@ unsigned long NormSlidingMask::NextSet(unsigned long index) const
pos = (maskIndex << 3) + loc; pos = (maskIndex << 3) + loc;
pos -= start; pos -= start;
if (pos < 0) pos += num_bits; 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 = (maskIndex << 3) + BITLOCS[mask[maskIndex]][0];
pos -= start; pos -= start;
if (pos < 0) pos += num_bits; if (pos < 0) pos += num_bits;
return (offset + pos); index = offset + pos;
return true;
} }
} }
maskIndex = 0; maskIndex = 0;
@ -968,36 +1010,41 @@ unsigned long NormSlidingMask::NextSet(unsigned long index) const
pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][0]; pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][0];
pos -= start; pos -= start;
if (pos < 0) pos += num_bits; if (pos < 0) pos += num_bits;
return (offset + pos); index = offset + pos;
return true;
} }
} }
} }
else else
{ {
return offset; return false;
} }
} }
return (offset + num_bits); // indicates nothing was set return false; // indicates nothing was set
} // end NormSlidingMask::NextSet() } // end NormSlidingMask::GetNextSet()
unsigned long NormSlidingMask::PrevSet(unsigned long index) const bool NormSlidingMask::GetPrevSet(unsigned long& index) const
{ {
if (IsSet()) if (IsSet())
{ {
long pos = index - offset; unsigned long prev = index;
long pos = Delta(prev, offset);
if (pos >= 0) if (pos >= 0)
{ {
// Is it in range? // Is it in range?
if (pos >= num_bits) return (offset + num_bits); if (pos >= num_bits)
{
return false;
}
pos += start; pos += start;
if (pos >= num_bits) pos -= num_bits; if (pos >= num_bits) pos -= num_bits;
if (end < start) if (end < start)
{ {
if ((pos > end) && (pos < start)) return (offset + num_bits); if ((pos > end) && (pos < start)) return false;
} }
else else
{ {
if ((pos < start) || (pos > end)) return (offset + num_bits); if ((pos < start) || (pos > end)) return false;
} }
// Seek prev set bits, starting with index // Seek prev set bits, starting with index
long maskIndex = pos >> 3; long maskIndex = pos >> 3;
@ -1013,7 +1060,8 @@ unsigned long NormSlidingMask::PrevSet(unsigned long index) const
pos = (maskIndex << 3) + loc; pos = (maskIndex << 3) + loc;
pos -= start; pos -= start;
if (pos < 0) pos += num_bits; 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 = (maskIndex << 3) + BITLOCS[mask[maskIndex]][w];
pos -= start; pos -= start;
if (pos < 0) pos += num_bits; if (pos < 0) pos += num_bits;
return (offset + pos); index = offset + pos;
return true;
} }
} }
maskIndex = mask_len - 1; maskIndex = mask_len - 1;
@ -1043,15 +1092,16 @@ unsigned long NormSlidingMask::PrevSet(unsigned long index) const
pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][w]; pos = (maskIndex << 3) + BITLOCS[mask[maskIndex]][w];
pos -= start; pos -= start;
if (pos < 0) pos += num_bits; if (pos < 0) pos += num_bits;
return (offset + pos); index = offset + pos;
return true;
} }
} }
} }
} }
return (offset + num_bits); // indicates nothing was set return false; // indicates nothing prior was set
} // end NormSlidingMask::PrevSet() } // 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 // Seek next set bit
unsigned long maskIndex = index >> 3; 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) return (index+1); // (nothing was set)
} // end NormSlidingMask::RawPrevSet() } // end NormSlidingMask::RawPrevSet()*/
bool NormSlidingMask::Copy(const NormSlidingMask& b) bool NormSlidingMask::Copy(const NormSlidingMask& b)
{ {
@ -1115,7 +1165,11 @@ bool NormSlidingMask::Copy(const NormSlidingMask& b)
if (range <= num_bits) if (range <= num_bits)
{ {
start = b.start & 0x07; 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; offset = b.offset;
// Copy start to mask_len // Copy start to mask_len
long startIndex = b.start >> 3; long startIndex = b.start >> 3;
@ -1166,11 +1220,16 @@ bool NormSlidingMask::Add(const NormSlidingMask& b)
{ {
if (IsSet()) if (IsSet())
{ {
if (!CanSet(b.FirstSet())) return false; UINT32 bFirstSet;
if (!CanSet(b.LastSet())) return false; b.GetFirstSet(bFirstSet);
if (!CanSet(bFirstSet)) return false;
UINT32 bLastSet;
b.GetFirstSet(bLastSet);
if (!CanSet(bLastSet)) return false;
long range = b.end - b.start; long range = b.end - b.start;
if (range < 0) range += b.num_bits; 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++) for (long i = 0; i < range; i++)
{ {
// (TBD) Improve performance by getting/setting // (TBD) Improve performance by getting/setting
@ -1206,10 +1265,6 @@ bool NormSlidingMask::Subtract(const NormSlidingMask& b)
} }
} }
} }
else
{
Copy(b);
}
return true; return true;
} // end NormSlidingMask::Subtract() } // end NormSlidingMask::Subtract()
@ -1221,9 +1276,14 @@ bool NormSlidingMask::XCopy(const NormSlidingMask& b)
{ {
if (IsSet()) if (IsSet())
{ {
if (!CanSet(b.FirstSet())) return false; UINT32 bFirstSet;
if (!CanSet(b.LastSet())) return false; b.GetFirstSet(bFirstSet);
unsigned long index = b.FirstSet(); 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; long range = b.end - b.start;
if (range < 0) range += b.num_bits; if (range < 0) range += b.num_bits;
for (long i = 0; i < range; i++) for (long i = 0; i < range; i++)
@ -1237,9 +1297,13 @@ bool NormSlidingMask::XCopy(const NormSlidingMask& b)
} }
else else
{ {
Copy(b); return Copy(b);
} }
} }
else
{
Clear();
}
return true; return true;
} // end NormSlidingMask::XCopy() } // end NormSlidingMask::XCopy()
@ -1268,13 +1332,19 @@ bool NormSlidingMask::Multiply(const NormSlidingMask& b)
} // end NormSlidingMask::Multiply() } // end NormSlidingMask::Multiply()
// Logically XOR two bit mask such that "this = (this ^ b)" // 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) bool NormSlidingMask::Xor(const NormSlidingMask& b)
{ {
if (b.IsSet()) if (b.IsSet())
{ {
if (!CanSet(b.FirstSet())) return false; UINT32 bFirstSet;
if (!CanSet(b.LastSet())) return false; b.GetFirstSet(bFirstSet);
unsigned long index = b.FirstSet(); 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; long range = b.end - b.start;
if (range < 0) range += b.num_bits; if (range < 0) range += b.num_bits;
for (long i = 0; i < range; i++) 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 ");
} }
if (0x3f != (i & 0x3f)) DMSG(0, "\n"); if (0x3f != (i & 0x3f)) DMSG(0, "\n");
} // end NormSlidingMask::Display() } // end NormSlidingMask::Debug()

59
common/normBitmask.h
View File

@ -37,11 +37,16 @@ class NormBitmask
} }
bool IsSet() const {return (first_set < num_bits);} bool IsSet() const {return (first_set < num_bits);}
unsigned long FirstSet() const {return first_set;} bool GetFirstSet(unsigned long& index) const
unsigned long LastSet() const
{ {
return ((first_set < num_bits) ? index = first_set;
PrevSet(num_bits - 1) : num_bits); return IsSet();
}
bool GetLastSet(unsigned long& index) const
{
index = num_bits - 1;
return GetPrevSet(index);
} }
bool Test(unsigned long index) const bool Test(unsigned long index) const
{ {
@ -70,19 +75,19 @@ class NormBitmask
if (index < num_bits) if (index < num_bits)
{ {
mask[(index >> 3)] &= ~(0x80 >> (index & 0x07)); 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; return true;
} }
bool Invert(unsigned long index) bool Invert(unsigned long index)
{return (Test(index) ? Unset(index) : Set(index));} {return (Test(index) ? Unset(index) : Set(index));}
bool SetBits(unsigned long baseIndex, unsigned long count); bool SetBits(unsigned long baseIndex, unsigned long count);
bool UnsetBits(unsigned long baseIndex, unsigned long count); bool UnsetBits(unsigned long baseIndex, unsigned long count);
unsigned long NextSet(unsigned long index) const; bool GetNextSet(unsigned long& index) const;
unsigned long PrevSet(unsigned long index) const; bool GetPrevSet(unsigned long& index) const;
unsigned long NextUnset(unsigned long index) const; bool GetNextUnset(unsigned long& index) const;
bool Copy(const NormBitmask &b); // this = b bool Copy(const NormBitmask &b); // this = b
bool Add(const NormBitmask & b); // this = this | b bool Add(const NormBitmask & b); // this = this | b
@ -120,7 +125,7 @@ class NormSlidingMask
const char* GetMask() const {return (const char*)mask;} const char* GetMask() const {return (const char*)mask;}
bool Init(long numBits); bool Init(long numBits, UINT32 rangeMax);
void Destroy(); void Destroy();
long Size() const {return num_bits;} long Size() const {return num_bits;}
void Clear() void Clear()
@ -139,12 +144,17 @@ class NormSlidingMask
offset = index; offset = index;
} }
bool IsSet() const {return (start < num_bits);} bool IsSet() const {return (start < num_bits);}
unsigned long FirstSet() const {return offset;} bool GetFirstSet(unsigned long& index) const
unsigned long LastSet() const {
index = offset;
return IsSet();
}
bool GetLastSet(unsigned long& index) const
{ {
long n = end - start; long n = end - start;
n = (n < 0) ? (n + num_bits) : n; n = (n < 0) ? (n + num_bits) : n;
return (n + offset); index = offset + n;
return IsSet();
} }
bool Test(unsigned long index) const; bool Test(unsigned long index) const;
bool CanSet(unsigned long index) const; bool CanSet(unsigned long index) const;
@ -159,12 +169,12 @@ class NormSlidingMask
// These return "FirstSet()+Size()" when finding nothing // These return "false" when finding nothing
unsigned long NextSet(unsigned long index) const; bool GetNextSet(unsigned long& index) const;
unsigned long PrevSet(unsigned long index) const; bool GetPrevSet(unsigned long& index) const;
static unsigned long RawNextSet(const char* mask, long index, long start); //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 RawPrevSet(const char* mask, long index, long end);
bool Copy(const NormSlidingMask& b); // this = b bool Copy(const NormSlidingMask& b); // this = b
@ -178,8 +188,21 @@ class NormSlidingMask
void Debug(long theCount); void Debug(long theCount);
private: 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 char* mask;
unsigned long mask_len; unsigned long mask_len;
unsigned long range_mask;
long range_sign;
long num_bits; long num_bits;
long start; long start;
long end; long end;

20
common/normEncoder.cpp
View File

@ -62,6 +62,11 @@ bool NormEncoder::Init(int numParity, int vecSizeMax)
ASSERT(vecSizeMax >= 0); ASSERT(vecSizeMax >= 0);
if (genPoly) Destroy(); if (genPoly) Destroy();
npar = numParity; npar = numParity;
#ifdef SIMULATE
vecSizeMax = MIN(SIM_PAYLOAD_MAX+1, vecSizeMax);
#endif // SIMUATE
vector_size = vecSizeMax; vector_size = vecSizeMax;
// Create generator polynomial // Create generator polynomial
if(!CreateGeneratorPolynomial()) if(!CreateGeneratorPolynomial())
@ -180,7 +185,7 @@ void NormEncoder::Encode(const char *data, char **pVec)
// Copy pVec[0] for use in calculations // Copy pVec[0] for use in calculations
#ifdef SIMULATE #ifdef SIMULATE
UINT16 vecSize = MIN(SIM_PAYLOAD_MAX, vector_size); UINT16 vecSize = MIN(SIM_PAYLOAD_MAX+1, vector_size);
#else #else
UINT16 vecSize = vector_size; UINT16 vecSize = vector_size;
#endif // if/else SIMULATE #endif // if/else SIMULATE
@ -230,6 +235,10 @@ bool NormDecoder::Init(int numParity, int vecSizeMax)
ASSERT((numParity>=0)&&(numParity<=128)); ASSERT((numParity>=0)&&(numParity<=128));
ASSERT(vecSizeMax >= 0); ASSERT(vecSizeMax >= 0);
#ifdef SIMULATE
vecSizeMax = MIN(SIM_PAYLOAD_MAX+1, vecSizeMax);
#endif // SIMUATE
if (Lambda) Destroy(); // Check if already inited ... if (Lambda) Destroy(); // Check if already inited ...
npar = numParity; npar = numParity;
@ -328,8 +337,7 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era
// Debugging assertions // Debugging assertions
ASSERT(Lambda); ASSERT(Lambda);
ASSERT(erasureCount && (erasureCount<=npar)); ASSERT(erasureCount && (erasureCount<=npar));
// (A) Compute syndrome vectors // (A) Compute syndrome vectors
// First zero out erasure vectors (MDP provides zero-filled vecs) // 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) // Then calculate syndrome (based on zero value erasures)
int nvecs = npar + ndata; int nvecs = npar + ndata;
#ifdef SIMULATE #ifdef SIMULATE
int vecSize = MIN(SIM_PAYLOAD_MAX, vector_size); int vecSize = MIN(SIM_PAYLOAD_MAX+1, vector_size);
#else #else
int vecSize = vector_size; int vecSize = vector_size;
#endif // if/else SIMUATE #endif // if/else SIMUATE
@ -347,8 +355,8 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era
{ {
int X = gexp(i+1); int X = gexp(i+1);
unsigned char* synVec = sVec[i]; unsigned char* synVec = sVec[i];
memset(synVec, 0, vecSize*sizeof(char)); memset(synVec, 0, vecSize*sizeof(char));
for(int j = 0; j < nvecs; j++) for(int j = 0; j < nvecs; j++)
{ {
unsigned char* data = dVec[j] ? (unsigned char*)dVec[j] : scratch; unsigned char* data = dVec[j] ? (unsigned char*)dVec[j] : scratch;
unsigned char* S = synVec; unsigned char* S = synVec;

85
common/normMessage.cpp
View File

@ -425,30 +425,6 @@ unsigned char NormQuantizeRtt(double rtt)
return ((unsigned char)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt))))); return ((unsigned char)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt)))));
} // end NormQuantizeRtt() } // 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 NormObjectSize::operator+(const NormObjectSize& size) const
{ {
NormObjectSize total; NormObjectSize total;
@ -557,64 +533,3 @@ NormObjectSize NormObjectSize::operator/(const NormObjectSize& b) const
return result; return result;
} // end NormObjectSize::operator/(NormObjectSize b) } // 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

47
common/normMessage.h
View File

@ -106,15 +106,24 @@ class NormObjectSize
UINT16 MSB() const {return msb;} UINT16 MSB() const {return msb;}
UINT32 LSB() const {return lsb;} UINT32 LSB() const {return lsb;}
bool operator<(const NormObjectSize& size) const;
bool operator<=(const NormObjectSize& size) const
{return ((*this == size) || (*this<size));}
bool operator>(const NormObjectSize& size) const;
bool operator>=(const NormObjectSize& size) const
{return ((*this == size) || (*this>size));}
bool operator==(const NormObjectSize& size) const bool operator==(const NormObjectSize& size) const
{return ((msb == size.msb) && (lsb == size.lsb));} {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<size));}
bool operator>=(const NormObjectSize& size) const
{return ((*this == size) || (*this>size));}
bool operator!=(const NormObjectSize& size) const bool operator!=(const NormObjectSize& size) const
{return ((lsb != size.lsb) || (msb != size.msb));} {return ((lsb != size.lsb) || (msb != size.msb));}
NormObjectSize operator+(const NormObjectSize& size) const; NormObjectSize operator+(const NormObjectSize& size) const;
@ -158,17 +167,26 @@ class NormObjectId
NormObjectId(UINT16 id) {value = id;} NormObjectId(UINT16 id) {value = id;}
NormObjectId(const NormObjectId& id) {value = id.value;} NormObjectId(const NormObjectId& id) {value = id.value;}
operator UINT16() const {return 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 bool operator<=(const NormObjectId& id) const
{return ((value == id.value) || (*this<id));} {return ((value == id.value) || (*this<id));}
bool operator>(const NormObjectId& id) const;
bool operator>=(const NormObjectId& id) const bool operator>=(const NormObjectId& id) const
{return ((value == id.value) || (*this>id));} {return ((value == id.value) || (*this>id));}
bool operator==(const NormObjectId& id) const bool operator==(const NormObjectId& id) const
{return (value == id.value);} {return (value == id.value);}
bool operator!=(const NormObjectId& id) const bool operator!=(const NormObjectId& id) const
{return (value != id.value);} {return (value != id.value);}
int operator-(const NormObjectId& id) const;
NormObjectId& operator++(int ) {value++; return *this;} NormObjectId& operator++(int ) {value++; return *this;}
private: private:
@ -182,13 +200,16 @@ class NormBlockId
NormBlockId(UINT32 id) {value = id;} NormBlockId(UINT32 id) {value = id;}
NormBlockId(const NormObjectId& id) {value = (UINT32)id;} NormBlockId(const NormObjectId& id) {value = (UINT32)id;}
operator UINT32() const {return value;} operator UINT32() const {return value;}
bool operator<(const NormBlockId& id) const;
bool operator>(const NormBlockId& id) const;
bool operator==(const NormBlockId& id) const bool operator==(const NormBlockId& id) const
{return (value == (UINT32)id);} {return (value == (UINT32)id);}
bool operator!=(const NormBlockId& id) const bool operator!=(const NormBlockId& id) const
{return (value != (UINT32)id);} {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;} NormBlockId& operator++(int ) {value++; return *this;}
private: private:

160
common/normNode.cpp
View File

@ -82,13 +82,13 @@ bool NormServerNode::Open(UINT16 sessionId, UINT16 segmentSize, UINT16 numData,
Close(); Close();
return false; 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"); DMSG(0, "NormServerNode::Open() rx_pending_mask init error\n");
Close(); Close();
return false; 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"); DMSG(0, "NormServerNode::Open() rx_repair_mask init error\n");
Close(); Close();
@ -271,7 +271,7 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
} }
else if (!cc_timer.IsActive()) else if (!cc_timer.IsActive())
{ {
double backoffFactor = session->BackoffFactor(); double backoffFactor = backoff_factor;
backoffFactor = MAX(backoffFactor, 4.0); backoffFactor = MAX(backoffFactor, 4.0);
double maxBackoff = grtt_estimate*backoffFactor; double maxBackoff = grtt_estimate*backoffFactor;
// (TBD) don't backoff for unicast sessions // (TBD) don't backoff for unicast sessions
@ -312,6 +312,7 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
case NormCmdMsg::FLUSH: case NormCmdMsg::FLUSH:
// (TBD) should we force synchronize if we're expected // (TBD) should we force synchronize if we're expected
// to positively acknowledge the FLUSH // to positively acknowledge the FLUSH
if (synchronized) if (synchronized)
{ {
const NormCmdFlushMsg& flush = (const NormCmdFlushMsg&)cmd; const NormCmdFlushMsg& flush = (const NormCmdFlushMsg&)cmd;
@ -397,7 +398,7 @@ void NormServerNode::HandleCCFeedback(UINT8 ccFlags, double ccRate)
if (suppressed) if (suppressed)
{ {
if (cc_timer.IsActive()) cc_timer.Deactivate(); 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); session->ActivateTimer(cc_timer);
cc_timer.DecrementRepeatCount(); cc_timer.DecrementRepeatCount();
} }
@ -910,11 +911,12 @@ void NormServerNode::Sync(NormObjectId objectId)
{ {
if (synchronized) if (synchronized)
{ {
if (rx_pending_mask.IsSet()) NormObjectId firstPending;
if (GetFirstPending(firstPending))
{ {
NormObjectId firstSet = NormObjectId((UINT16)rx_pending_mask.FirstSet()); NormObjectId lastPending;
if ((objectId > NormObjectId((UINT16)rx_pending_mask.LastSet())) || GetLastPending(lastPending);
((next_id - objectId) > 256)) if ((objectId > lastPending) || ((next_id - objectId) > 256))
{ {
NormObject* obj; NormObject* obj;
while ((obj = rx_table.Find(rx_table.RangeLo()))) while ((obj = rx_table.Find(rx_table.RangeLo())))
@ -924,7 +926,7 @@ void NormServerNode::Sync(NormObjectId objectId)
} }
rx_pending_mask.Clear(); rx_pending_mask.Clear();
} }
else if (objectId > firstSet) else if (objectId > firstPending)
{ {
NormObject* obj; NormObject* obj;
while ((obj = rx_table.Find(rx_table.RangeLo())) && while ((obj = rx_table.Find(rx_table.RangeLo())) &&
@ -933,12 +935,11 @@ void NormServerNode::Sync(NormObjectId objectId)
DeleteObject(obj); DeleteObject(obj);
failure_count++; failure_count++;
} }
unsigned long numBits = (UINT16)(objectId - firstSet) + 1; unsigned long numBits = (UINT16)(objectId - firstPending) + 1;
rx_pending_mask.UnsetBits(firstSet, numBits); rx_pending_mask.UnsetBits(firstPending, numBits);
} }
} }
if ((next_id < objectId) || if ((next_id < objectId) || ((next_id - objectId) > 256))
((next_id - objectId) > 256))
{ {
max_pending_object = next_id = objectId; 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 // (TBD) We may want to control re-sync policy options
// or revert to fresh sync if sync is totally lost, // or revert to fresh sync if sync is totally lost,
// otherwise SQUELCH process will get things in order // 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()); LocalNodeId(), GetId());
Sync(objectId); Sync(objectId);
resync_count++; resync_count++;
@ -990,7 +991,7 @@ void NormServerNode::SetPending(NormObjectId objectId)
rx_pending_mask.SetBits(next_id, objectId - next_id + 1); rx_pending_mask.SetBits(next_id, objectId - next_id + 1);
next_id = objectId + 1; next_id = objectId + 1;
// This prevents the "sync_id" from getting stale // This prevents the "sync_id" from getting stale
sync_id = (UINT16)rx_pending_mask.FirstSet(); GetFirstPending(sync_id);
} }
} // end NormServerNode::SetPending() } // end NormServerNode::SetPending()
@ -1068,19 +1069,18 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel,
if (!repair_timer.IsActive()) if (!repair_timer.IsActive())
{ {
bool startTimer = false; bool startTimer = false;
if (rx_pending_mask.IsSet()) NormObjectId nextId;
if (GetFirstPending(nextId))
{ {
if (rx_repair_mask.IsSet()) rx_repair_mask.Clear(); if (rx_repair_mask.IsSet()) rx_repair_mask.Clear();
NormObjectId nextId = (UINT16)rx_pending_mask.FirstSet(); do
NormObjectId lastId = (UINT16)rx_pending_mask.LastSet();
if (objectId < lastId) lastId = objectId;
while (nextId <= lastId)
{ {
if (nextId > objectId) break;
NormObject* obj = rx_table.Find(nextId); NormObject* obj = rx_table.Find(nextId);
if (obj) if (obj)
{ {
NormObject::CheckLevel level; NormObject::CheckLevel level;
if (nextId < lastId) if (nextId < objectId)
level = NormObject::THRU_OBJECT; level = NormObject::THRU_OBJECT;
else else
level = checkLevel; level = checkLevel;
@ -1092,18 +1092,17 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel,
startTimer = true; startTimer = true;
} }
nextId++; nextId++;
nextId = (UINT16)rx_pending_mask.NextSet(nextId); } while (GetNextPending(nextId));
}
current_object_id = objectId; current_object_id = objectId;
if (startTimer) if (startTimer)
{ {
// BACKOFF related code // BACKOFF related code
double backoffInterval = double backoffInterval =
(session->Address().IsMulticast() && (session->BackoffFactor() > 0.0)) ? (session->Address().IsMulticast() && (backoff_factor > 0.0)) ?
ExponentialRand(grtt_estimate*session->BackoffFactor(), gsize_estimate) : ExponentialRand(grtt_estimate*backoff_factor, gsize_estimate) :
0.0; 0.0;
repair_timer.SetInterval(backoffInterval); 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); LocalNodeId(), backoffInterval);
session->ActivateTimer(repair_timer); session->ActivateTimer(repair_timer);
} }
@ -1155,14 +1154,15 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu end NACK back-off ...\n", DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu end NACK back-off ...\n",
LocalNodeId()); LocalNodeId());
// 1) Were we suppressed? // 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; bool repairPending = false;
NormObjectId nextId = (UINT16)rx_pending_mask.FirstSet(); do
NormObjectId lastId = (UINT16)rx_pending_mask.LastSet();
if (current_object_id < lastId) lastId = current_object_id;
while (nextId <= lastId)
{ {
if (nextId > current_object_id) break;
if (!rx_repair_mask.Test(nextId)) if (!rx_repair_mask.Test(nextId))
{ {
NormObject* obj = rx_table.Find(nextId); NormObject* obj = rx_table.Find(nextId);
@ -1173,8 +1173,8 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
} }
} }
nextId++; nextId++;
nextId = (UINT16)rx_pending_mask.NextSet(nextId); } while (GetNextPending(nextId));
} // end while (nextId <= current_block_id)
if (repairPending) if (repairPending)
{ {
// We weren't completely suppressed, so build NACK // We weren't completely suppressed, so build NACK
@ -1222,38 +1222,34 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
if (cc_timer.IsActive()) if (cc_timer.IsActive())
{ {
cc_timer.Deactivate(); cc_timer.Deactivate();
cc_timer.SetInterval(grtt_estimate*session->BackoffFactor()); cc_timer.SetInterval(grtt_estimate*backoff_factor);
session->ActivateTimer(cc_timer); session->ActivateTimer(cc_timer);
cc_timer.DecrementRepeatCount(); cc_timer.DecrementRepeatCount();
} }
} }
// Iterate through rx pending object list, appending repair requests as needed
NormRepairRequest req; NormRepairRequest req;
NormObjectId prevId;
UINT16 reqCount = 0;
NormRepairRequest::Form prevForm = NormRepairRequest::INVALID; NormRepairRequest::Form prevForm = NormRepairRequest::INVALID;
nextId = (UINT16)rx_pending_mask.FirstSet(); bool iterating = GetFirstPending(nextId);
lastId = (UINT16)rx_pending_mask.LastSet(); iterating = iterating && (nextId <= max_pending_object);
if (max_pending_object < lastId) lastId = max_pending_object; NormObjectId prevId = nextId;
lastId++; // force loop to fully flush nack building. UINT16 consecutiveCount = 0;
while ((nextId <= lastId) || (reqCount > 0)) while (iterating || (0 != consecutiveCount))
{ {
NormObject* obj = NULL; bool appendRequest = false;
bool objPending = false; NormObject* obj = iterating ? rx_table.Find(nextId) : NULL;
if (nextId == lastId) if (obj)
nextId++; // force break of possible ending consecutive series appendRequest = true;
else if (iterating && ((nextId - prevId) == consecutiveCount))
consecutiveCount++; // consecutive series of missing objs starts/continues
else else
obj = rx_table.Find(nextId); appendRequest = true; // consecutive series broken or finished
if (obj) objPending = obj->IsPending(nextId != max_pending_object);
if (!objPending && reqCount && (reqCount == (nextId - prevId))) if (appendRequest)
{
reqCount++; // consecutive series of missing objs continues
}
else
{ {
NormRepairRequest::Form nextForm; NormRepairRequest::Form nextForm;
switch (reqCount) switch (consecutiveCount)
{ {
case 0: case 0:
nextForm = NormRepairRequest::INVALID; nextForm = NormRepairRequest::INVALID;
@ -1265,7 +1261,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
default: default:
nextForm = NormRepairRequest::RANGES; nextForm = NormRepairRequest::RANGES;
break; break;
} }
if (prevForm != nextForm) if (prevForm != nextForm)
{ {
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
@ -1280,41 +1276,47 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
} }
prevForm = nextForm; prevForm = nextForm;
} }
if (NormRepairRequest::INVALID != nextForm)
DMSG(6, "NormServerNode::AppendRepairRequest() OBJECT request\n");
switch (nextForm) switch (nextForm)
{ {
case NormRepairRequest::ITEMS: case NormRepairRequest::ITEMS:
req.AppendRepairItem(prevId, 0, ndata, 0); req.AppendRepairItem(prevId, 0, ndata, 0);
if (2 == reqCount) if (2 == consecutiveCount)
req.AppendRepairItem(prevId+1, 0, ndata, 0); req.AppendRepairItem(prevId+1, 0, ndata, 0);
break; break;
case NormRepairRequest::RANGES: case NormRepairRequest::RANGES:
req.AppendRepairItem(prevId, 0, ndata, 0); req.AppendRepairRange(prevId, 0, ndata, 0,
req.AppendRepairItem(prevId+reqCount-1, 0, ndata, 0); prevId+consecutiveCount-1, 0, ndata, 0);
break; break;
default: default:
break; break;
} }
prevId = nextId;
if (obj || (nextId >= lastId)) if (obj)
reqCount = 0; {
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 else
reqCount = 1; {
} // end if/else (!objPending && reqCount && (reqCount == (nextId - prevId))) consecutiveCount = 0; // we're all done
if (objPending) }
{ prevId = nextId;
if (NormRepairRequest::INVALID != prevForm) } // end if (appendRequest)
nack->PackRepairRequest(req); // (TBD) error check
prevForm = NormRepairRequest::INVALID;
reqCount = 0;
bool flush = (nextId != max_pending_object);
obj->AppendRepairRequest(*nack, flush); // (TBD) error check
}
nextId++; nextId++;
if (nextId <= lastId) iterating = GetNextPending(nextId);
nextId = (UINT16)rx_pending_mask.NextSet(nextId); } while (iterating || (0 != consecutiveCount));
} // end while(nextId <= lastId) // Now that all repair requests have been appended, pack the nack
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
nack->PackRepairRequest(req); // (TBD) error check nack->PackRepairRequest(req); // (TBD) error check
// Queue NACK for transmission // Queue NACK for transmission
@ -1338,9 +1340,9 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
} // end if/else(repairPending) } // end if/else(repairPending)
// BACKOFF related code // BACKOFF related code
double holdoffInterval = double holdoffInterval =
session->Address().IsMulticast() ? grtt_estimate*(session->BackoffFactor() + 2.0) : session->Address().IsMulticast() ? grtt_estimate*(backoff_factor + 2.0) :
grtt_estimate; 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); repair_timer.SetInterval(holdoffInterval);
DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu begin NACK hold-off: %lf sec ...\n", 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 // Begin cc_timer "holdoff" phase
cc_timer.SetInterval(grtt_estimate*session->BackoffFactor()); cc_timer.SetInterval(grtt_estimate*backoff_factor);
return true; return true;
} }

26
common/normNode.h
View File

@ -209,9 +209,31 @@ class NormServerNode : public NormNode
bool SyncTest(const NormObjectMsg& msg) const; bool SyncTest(const NormObjectMsg& msg) const;
void Sync(NormObjectId objectId); void Sync(NormObjectId objectId);
ObjectStatus UpdateSyncStatus(const NormObjectId& objectId); ObjectStatus UpdateSyncStatus(const NormObjectId& objectId);
void SetPending(NormObjectId objectId);
ObjectStatus GetObjectStatus(const NormObjectId& objectId) const; 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); void DeleteObject(NormObject* obj);
UINT16 SegmentSize() {return segment_size;} UINT16 SegmentSize() {return segment_size;}
@ -237,7 +259,7 @@ class NormServerNode : public NormNode
{return erasure_loc[index];} {return erasure_loc[index];}
UINT16 Decode(char** segmentList, UINT16 numData, UINT16 erasureCount) 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, void CalculateGrttResponse(const struct timeval& currentTime,

399
common/normObject.cpp
View File

@ -102,7 +102,7 @@ bool NormObject::Open(const NormObjectSize& objectSize,
} }
// Init pending_mask (everything pending) // 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"); DMSG(0, "NormObject::Open() init pending_mask error\n");
Close(); Close();
@ -112,7 +112,7 @@ bool NormObject::Open(const NormObjectSize& objectSize,
pending_mask.SetBits(0, numBlocks.LSB()); pending_mask.SetBits(0, numBlocks.LSB());
// Init repair_mask // 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"); DMSG(0, "NormObject::Open() init pending_mask error\n");
Close(); Close();
@ -152,8 +152,8 @@ bool NormObject::Open(const NormObjectSize& objectSize,
ASSERT(0 == smallBlockCount.MSB()); ASSERT(0 == smallBlockCount.MSB());
small_block_count = smallBlockCount.LSB(); small_block_count = smallBlockCount.LSB();
} }
// not used for STREAM objects
final_block_id = large_block_count + small_block_count - 1; // not used for STREAM objects final_block_id = large_block_count + small_block_count - 1;
NormObjectSize finalSegmentSize = NormObjectSize finalSegmentSize =
objectSize - (numSegments - NormObjectSize((UINT32)1))*segmentSize; objectSize - (numSegments - NormObjectSize((UINT32)1))*segmentSize;
ASSERT(0 == finalSegmentSize.MSB()); ASSERT(0 == finalSegmentSize.MSB());
@ -291,26 +291,24 @@ bool NormObject::ActivateRepairs()
repairsActivated = true; repairsActivated = true;
} }
// Activate any complete block repairs // 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; 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(); UINT16 autoParity = session->ServerAutoParity();
while (nextId <= lastId) do
{ {
NormBlock* block = block_buffer.Find(nextId); NormBlock* block = block_buffer.Find(nextId);
if (block) block->TxReset(GetBlockSize(nextId), nparity, autoParity, segment_size); if (block) block->TxReset(GetBlockSize(nextId), nparity, autoParity, segment_size);
// (TBD) This check can be eventually eliminated if everything else is done right // (TBD) This check can be eventually eliminated if everything else is done right
if (!pending_mask.Set(nextId)) if (!pending_mask.Set(nextId))
DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error!\n", DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error!\n", (UINT32)nextId);
(UINT32)nextId); nextId++;
nextId++; } while (GetNextRepair(nextId));
nextId = repair_mask.NextSet(nextId);
}
repair_mask.Clear(); repair_mask.Clear();
} }
// Activate partial block (segment) repairs // Activate partial block (segment) repairs
@ -335,8 +333,10 @@ bool NormObject::ActivateRepairs()
bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd) bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd)
{ {
// Determine range of blocks possibly pending repair // Determine range of blocks possibly pending repair
NormBlockId nextId = repair_mask.FirstSet(); NormBlockId nextId;
NormBlockId endId = repair_mask.LastSet(); GetFirstRepair(nextId);
NormBlockId endId;
GetLastRepair(endId);
if (block_buffer.IsEmpty()) if (block_buffer.IsEmpty())
{ {
if (repair_mask.IsSet()) endId++; if (repair_mask.IsSet()) endId++;
@ -466,9 +466,9 @@ bool NormObject::IsPending(bool flush) const
} }
else else
{ {
if (pending_mask.IsSet()) NormBlockId firstId;
if (GetFirstPending(firstId))
{ {
NormBlockId firstId = pending_mask.FirstSet();
if (firstId < max_pending_block) if (firstId < max_pending_block)
{ {
return true; return true;
@ -484,7 +484,10 @@ bool NormObject::IsPending(bool flush) const
NormBlock* block = block_buffer.Find(max_pending_block); NormBlock* block = block_buffer.Find(max_pending_block);
if (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; return true;
else else
return false; return false;
@ -552,6 +555,7 @@ bool NormObject::ClientRepairCheck(CheckLevel level,
else if (blockId > max_pending_block) else if (blockId > max_pending_block)
max_pending_block = blockId; max_pending_block = blockId;
max_pending_segment = GetBlockSize(max_pending_block); max_pending_segment = GetBlockSize(max_pending_block);
break;
default: default:
break; break;
} // end switch (level) } // end switch (level)
@ -605,37 +609,43 @@ bool NormObject::ClientRepairCheck(CheckLevel level,
if ((level > THRU_INFO) && pending_mask.IsSet()) if ((level > THRU_INFO) && pending_mask.IsSet())
{ {
if (repair_mask.IsSet()) repair_mask.Clear(); if (repair_mask.IsSet()) repair_mask.Clear();
NormBlockId nextId = pending_mask.FirstSet(); NormBlockId nextId;
NormBlockId lastId = pending_mask.LastSet(); if (GetFirstPending(nextId))
if ((level < THRU_OBJECT) && (blockId < lastId)) {
lastId = blockId; NormBlockId lastId;
if (level > TO_BLOCK) lastId++; GetLastPending(lastId);
while (nextId < lastId) if ((level < THRU_OBJECT) && (blockId < lastId)) lastId = blockId;
{ if (level > TO_BLOCK) lastId++;
NormBlock* block = block_buffer.Find(nextId); while (nextId < lastId)
if (block)
{ {
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(); block->ClearRepairs();
needRepair = true; needRepair = true;
} }
} }
else else
{ {
block->ClearRepairs();
needRepair = true; needRepair = true;
} }
} nextId++;
else if (!GetNextPending(nextId)) break;
{ } // end while (nextId < lastId)
needRepair = true; }
}
nextId++;
nextId = pending_mask.NextSet(nextId);
}
} }
switch (level) switch (level)
{ {
@ -674,19 +684,18 @@ bool NormObject::IsRepairPending(bool flush)
if (pending_info && !repair_info) return true; if (pending_info && !repair_info) return true;
// Calculate repair_mask = pending_mask - repair_mask // Calculate repair_mask = pending_mask - repair_mask
repair_mask.XCopy(pending_mask); repair_mask.XCopy(pending_mask);
if (repair_mask.IsSet()) NormBlockId nextId;
if (GetFirstRepair(nextId))
{ {
NormBlockId nextId = repair_mask.FirstSet(); do
NormBlockId lastId = repair_mask.LastSet();
if (!flush && (current_block_id < lastId)) lastId = current_block_id;
while (nextId <= lastId)
{ {
if (!flush && (nextId > current_block_id)) break;
NormBlock* block = block_buffer.Find(nextId); NormBlock* block = block_buffer.Find(nextId);
if (block) if (block)
{ {
bool isPending; bool isPending;
UINT16 numData = GetBlockSize(nextId); UINT16 numData = GetBlockSize(nextId);
if (flush || (nextId < lastId)) if (flush || (nextId < current_block_id))
{ {
isPending = block->IsRepairPending(numData, nparity); isPending = block->IsRepairPending(numData, nparity);
} }
@ -704,8 +713,7 @@ bool NormObject::IsRepairPending(bool flush)
return true; // We need the whole thing return true; // We need the whole thing
} }
nextId++; nextId++;
nextId = repair_mask.NextSet(nextId); } while (GetNextRepair(nextId));
}
} }
return false; return false;
} // end NormObject::IsRepairPending() } // 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. // If !flush, we request only up _to_ max_pending_block::max_pending_segment.
NormRepairRequest req; NormRepairRequest req;
NormRepairRequest::Form prevForm = NormRepairRequest::INVALID; NormRepairRequest::Form prevForm = NormRepairRequest::INVALID;
NormBlockId prevId; NormBlockId nextId;
UINT16 reqCount = 0; bool iterating = GetFirstPending(nextId);
if (pending_mask.IsSet()) if (iterating)
{ {
NormBlockId nextId = pending_mask.FirstSet(); iterating = iterating && (flush || (nextId <= max_pending_block));
NormBlockId lastId = pending_mask.LastSet(); NormBlockId prevId = nextId;
if (!flush && (max_pending_block < lastId)) lastId = max_pending_block; UINT32 consecutiveCount = 0;
lastId++; while (iterating || (0 != consecutiveCount))
DMSG(6, "NormObject::AppendRepairRequest() node>%lu obj>%hu, blk>%lu->%lu (maxPending:%lu)\n", {
LocalNodeId(), (UINT16)id, NormBlockId lastId;
(UINT32)nextId, (UINT32)lastId, (UINT32)max_pending_block); ASSERT(GetLastPending(lastId));
DMSG(6, "NormObject::AppendRepairRequest() node>%lu obj>%hu, blk>%lu->%lu (maxPending:%lu)\n",
// (TBD) simplify this loop code LocalNodeId(), (UINT16)id,
while ((nextId <= lastId) || (reqCount > 0)) (UINT32)nextId, (UINT32)lastId, (UINT32)max_pending_block);
{ bool appendRequest = false;
NormBlock* block = NULL; NormBlock* block = iterating ? block_buffer.Find(nextId) : NULL;
bool blockPending = false;
if (nextId == lastId)
nextId++; // force break of possible ending consec. series
else
block = block_buffer.Find(nextId);
if (block) if (block)
appendRequest = true;
else if (iterating && ((UINT32)(nextId - prevId) == consecutiveCount))
{ {
if (nextId == max_pending_block) consecutiveCount++;
{
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
} }
else else
appendRequest = true;
if (appendRequest)
{ {
NormRepairRequest::Form nextForm; NormRepairRequest::Form nextForm;
switch(reqCount) switch(consecutiveCount)
{ {
case 0: case 0:
nextForm = NormRepairRequest::INVALID; nextForm = NormRepairRequest::INVALID;
@ -789,64 +783,89 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
switch (nextForm) switch (nextForm)
{ {
case NormRepairRequest::ITEMS: case NormRepairRequest::ITEMS:
req.AppendRepairItem(id, prevId, ndata, 0); // (TBD) error check req.AppendRepairItem(id, prevId, GetBlockSize(prevId), 0); // (TBD) error check
if (2 == reqCount) if (2 == consecutiveCount)
req.AppendRepairItem(id, prevId+1, ndata, 0); // (TBD) error check {
prevId++;
req.AppendRepairItem(id, prevId, GetBlockSize(prevId), 0); // (TBD) error check
}
break; break;
case NormRepairRequest::RANGES: 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; break;
}
default: default:
break; break;
} // end switch(nextForm) } // end switch(nextForm)
prevId = nextId; if (block)
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))
{ {
block->AppendRepairRequest(nack, numData, nparity, id, bool blockPending = false;
pending_info, segment_size); // (TBD) error check if (nextId == max_pending_block)
} {
else NormSymbolId firstPending = 0;
{ if (!block->GetFirstPending(firstPending)) ASSERT(0);
if (max_pending_segment < numData) if (firstPending < max_pending_segment) blockPending = true;
block->AppendRepairRequest(nack, max_pending_segment, 0, id, }
pending_info, segment_size); // (TBD) error check
else 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++; nextId++;
if (nextId <= lastId) iterating = GetNextPending(nextId);
nextId = pending_mask.NextSet(nextId); //TRACE("got next pending>%lu result:%d\n", (UINT32)nextId, iterating);
} // end while (nextId <= lastId) iterating = iterating && (flush || (nextId <= max_pending_block));
if (NormRepairRequest::INVALID != prevForm) } // end while (iterating || (0 != consecutiveCount))
nack.PackRepairRequest(req); // (TBD) error check //TRACE("bailed ...\n");
} }
else else
{ {
// INFO_ONLY repair request // INFO_ONLY repair request
ASSERT(pending_info); ASSERT(pending_info);
nack.AttachRepairRequest(req, segment_size); nack.AttachRepairRequest(req, segment_size);
prevForm = NormRepairRequest::ITEMS;
req.SetForm(NormRepairRequest::ITEMS); req.SetForm(NormRepairRequest::ITEMS);
req.ResetFlags(); req.ResetFlags();
req.SetFlag(NormRepairRequest::INFO); req.SetFlag(NormRepairRequest::INFO);
req.AppendRepairItem(id, 0, ndata, 0); // (TBD) error check req.AppendRepairItem(id, 0, 0, 0); // (TBD) error check
nack.PackRepairRequest(req); // (TBD) error check } // end if/else (iterating)
} // end if/else pending_mask.IsSet() if (NormRepairRequest::INVALID != prevForm)
nack.PackRepairRequest(req); // (TBD) error check
return true; return true;
} // end NormObject::AppendRepairRequest() } // end NormObject::AppendRepairRequest()
@ -900,9 +919,9 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
while (!stream->StreamUpdateStatus(blockId)) while (!stream->StreamUpdateStatus(blockId))
{ {
// Server is too far ahead of me ... // 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); NormBlock* block = block_buffer.Find(firstId);
if (block) if (block)
{ {
@ -959,10 +978,11 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
#ifdef SIMULATE #ifdef SIMULATE
// For simulations, we may need to cap the payloadLength // For simulations, we may need to cap the payloadLength
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
payloadLength = MIN(payloadLength, SIM_PAYLOAD_MAX);
#endif // SIMULATE #endif // SIMULATE
memcpy(segment, data.GetPayload(), payloadLength);
if (payloadLength < payloadMax) if (payloadLength < payloadMax)
memset(segment+payloadLength, 0, payloadMax-payloadLength); memset(segment+payloadLength, 0, payloadMax-payloadLength);
memcpy(segment, data.GetPayload(), payloadLength);
if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START))
segment[payloadMax] = NormObjectMsg::FLAG_MSG_START; segment[payloadMax] = NormObjectMsg::FLAG_MSG_START;
else else
@ -987,33 +1007,35 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
// and write any decoded data segments to object // and write any decoded data segments to object
DMSG(8, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu blk>%lu " DMSG(8, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu blk>%lu "
"completed block ...\n", LocalNodeId(), server->GetId(), "completed block ...\n", LocalNodeId(), server->GetId(),
(UINT16)id, (UINT32)block->Id()); (UINT16)id, (UINT32)block->Id());
UINT16 nextErasure = block->FirstPending(); UINT16 erasureCount = 0;
UINT16 erasureCount = 0; UINT16 nextErasure = 0;
if (nextErasure < numData) if (block->GetFirstPending(nextErasure))
{ {
UINT16 blockLen = numData + nparity; if (nextErasure < numData)
while (nextErasure < blockLen)
{ {
server->SetErasureLoc(erasureCount++, nextErasure); do
if (nextErasure < numData)
{ {
if (!(segment = server->GetFreeSegment(id, blockId))) server->SetErasureLoc(erasureCount++, nextErasure);
if (nextErasure < numData)
{ {
DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " if (!(segment = server->GetFreeSegment(id, blockId)))
"Warning! no free segments ...\n", LocalNodeId(), server->GetId(), {
(UINT16)id); DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
// (TBD) Dump the block ...??? "Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
return; (UINT16)id);
} // (TBD) Dump the block ...???
UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); return;
}
UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength();
#ifdef SIMULATE #ifdef SIMULATE
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
#endif // SIMULATE #endif // SIMULATE
memset(segment, 0, payloadMax+1); memset(segment, 0, payloadMax+1);
block->SetSegment(nextErasure, segment); block->SetSegment(nextErasure, segment);
} }
nextErasure = block->NextPending(nextErasure+1); nextErasure++;
} while (block->GetNextPending(nextErasure));
} }
} }
if (erasureCount) if (erasureCount)
@ -1188,11 +1210,10 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
pending_info = false; pending_info = false;
return true; return true;
} }
if (!pending_mask.IsSet()) return false; NormBlockId blockId;
if (!GetFirstPending(blockId)) return false;
NormDataMsg* data = (NormDataMsg*)msg; NormDataMsg* data = (NormDataMsg*)msg;
NormBlockId blockId = pending_mask.FirstSet();
UINT16 numData = GetBlockSize(blockId); UINT16 numData = GetBlockSize(blockId);
NormBlock* block = block_buffer.Find(blockId); NormBlock* block = block_buffer.Find(blockId);
if (!block) if (!block)
@ -1239,8 +1260,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
ASSERT(success); ASSERT(success);
} }
} }
ASSERT(block->IsPending()); NormSegmentId segmentId = 0;
NormSegmentId segmentId = block->FirstPending(); if (!block->GetFirstPending(segmentId)) ASSERT(0);
// Try to read segment // Try to read segment
if (segmentId < numData) if (segmentId < numData)
@ -1264,7 +1285,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
#ifdef SIMULATE #ifdef SIMULATE
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
#endif // SIMULATE #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 // Perform incremental FEC encoding as needed
@ -1596,7 +1617,6 @@ UINT16 NormFileObject::ReadSegment(NormBlockId blockId,
len = segment_size; len = segment_size;
} }
// Determine segment offset from blockId::segmentId // Determine segment offset from blockId::segmentId
NormObjectSize segmentOffset; NormObjectSize segmentOffset;
NormObjectSize segmentSize = NormObjectSize(segment_size); NormObjectSize segmentSize = NormObjectSize(segment_size);
@ -1800,7 +1820,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId)
// Handle potential sync block id wrap // Handle potential sync block id wrap
NormBlockId delta = stream_next_id - stream_sync_id; NormBlockId delta = stream_next_id - stream_sync_id;
if (delta > NormBlockId(2*pending_mask.Size())) if (delta > NormBlockId(2*pending_mask.Size()))
stream_sync_id = NormBlockId(pending_mask.FirstSet()); GetFirstPending(stream_sync_id);
return true; return true;
} }
else else
@ -1870,16 +1890,16 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId,
ASSERT(segment != NULL); ASSERT(segment != NULL);
UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment);
ASSERT(segmentLength <= segment_size); ASSERT(segmentLength <= segment_size);
#ifdef SIMULATE UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength();
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 payloadLength = segmentLength+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); memcpy(buffer, segment, payloadLength);
#endif // SIMULATE
buffer[payloadMax] = segment[payloadMax];
return payloadLength; return payloadLength;
} // end NormStreamObject::ReadSegment() } // end NormStreamObject::ReadSegment()
@ -1920,7 +1940,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
{ {
// Notify app for stream data salvage // Notify app for stream data salvage
read_index.block = block->Id(); read_index.block = block->Id();
read_index.segment = block->FirstPending(); block->GetFirstPending(read_index.segment);
NormBlock* tempBlock = block; NormBlock* tempBlock = block;
UINT32 tempOffset = read_offset; UINT32 tempOffset = read_offset;
session->Notify(NormController::RX_OBJECT_UPDATE, server, this); session->Notify(NormController::RX_OBJECT_UPDATE, server, this);
@ -1974,14 +1994,14 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
{ {
char* s = segment_pool.Get(); char* s = segment_pool.Get();
ASSERT(s != NULL); // for now, this should always succeed ASSERT(s != NULL); // for now, this should always succeed
UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); UINT16 payloadLength = NormDataMsg::ReadStreamPayloadLength(segment);
UINT16 flagsOffset = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength();
#ifdef SIMULATE #ifdef SIMULATE
flagsOffset = MIN(SIM_PAYLOAD_MAX, flagsOffset); payloadMax = MIN(SIM_PAYLOAD_MAX, payloadMax);
segmentLength = MIN((SIM_PAYLOAD_MAX-NormDataMsg::GetStreamPayloadHeaderLength()), segmentLength); payloadLength = MIN(payloadMax, payloadLength);
#endif // SIMULATE #endif // SIMULATE
s[flagsOffset] = segment[flagsOffset]; memcpy(s, segment, payloadLength);
memcpy(s, segment, segmentLength + NormDataMsg::GetStreamPayloadHeaderLength()); s[payloadMax] = segment[payloadMax];
block->AttachSegment(segmentId, s); block->AttachSegment(segmentId, s);
block->SetPending(segmentId); block->SetPending(segmentId);
ASSERT(block->Segment(segmentId) == s); ASSERT(block->Segment(segmentId) == s);
@ -2007,9 +2027,9 @@ void NormStreamObject::Prune(NormBlockId blockId)
break; break;
} }
} }
if (pending_mask.IsSet()) NormBlockId firstId;
if (GetFirstPending(firstId))
{ {
NormBlockId firstId = pending_mask.FirstSet();
if (firstId < blockId) if (firstId < blockId)
{ {
resync = true; resync = true;
@ -2079,11 +2099,11 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
} }
else else
{ {
UINT16 flagsOffset = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength();
#ifdef SIMULATE #ifdef SIMULATE
flagsOffset = MIN(flagsOffset, SIM_PAYLOAD_MAX); payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
#endif // if/else SIMULATE #endif // if/else SIMULATE
msgStart = (NormObjectMsg::FLAG_MSG_START == segment[flagsOffset]); msgStart = (NormObjectMsg::FLAG_MSG_START == segment[payloadMax]);
} }
if (!msgStart) if (!msgStart)
{ {
@ -2103,9 +2123,8 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
} }
#ifdef SIMULATE #ifdef SIMULATE
UINT16 simCount = ((index+count) < SIM_PAYLOAD_MAX) ? UINT16 simCount = index + count + NormDataMsg::GetStreamPayloadHeaderLength();
count : ((index < SIM_PAYLOAD_MAX) ? simCount = (simCount < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - simCount) : 0;
(SIM_PAYLOAD_MAX - index) : 0);
memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), simCount); memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), simCount);
#else #else
memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), count); memcpy(buffer+bytesRead, segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), count);
@ -2187,7 +2206,7 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len,
if (c) if (c)
{ {
block_buffer.Remove(c); block_buffer.Remove(c);
session->ServerPutFreeBlock(c); session->ServerPutFreeBlock(c);
} }
} }
else if (b->IsPending()) else if (b->IsPending())
@ -2217,11 +2236,11 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len,
msgStartValue = NormObjectMsg::FLAG_MSG_START; msgStartValue = NormObjectMsg::FLAG_MSG_START;
msg_start = false; msg_start = false;
} }
UINT16 flagsOffset = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength();
#ifdef SIMULATE #ifdef SIMULATE
flagsOffset = MIN(SIM_PAYLOAD_MAX, flagsOffset); payloadMax = MIN(SIM_PAYLOAD_MAX, payloadMax);
#endif // SIMULATE #endif // SIMULATE
segment[flagsOffset] = msgStartValue; segment[payloadMax] = msgStartValue;
} }
else else
{ {
@ -2232,11 +2251,13 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len,
UINT16 space = segment_size - index; UINT16 space = segment_size - index;
count = MIN(count, space); count = MIN(count, space);
#ifdef SIMULATE #ifdef SIMULATE
UINT16 simCount = ((index+count) < SIM_PAYLOAD_MAX) ? UINT16 simCount = index + NormDataMsg::GetStreamPayloadHeaderLength();
count : ((index < SIM_PAYLOAD_MAX) ? simCount = (simCount < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - simCount) : 0;
(SIM_PAYLOAD_MAX - index) : 0); simCount = MIN(count, simCount);
memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, simCount); memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, simCount);
#else #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); memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, count);
#endif // if/else SIMULATE #endif // if/else SIMULATE
NormDataMsg::WriteStreamPayloadLength(segment, index+count); NormDataMsg::WriteStreamPayloadLength(segment, index+count);
@ -2275,7 +2296,7 @@ unsigned long NormStreamObject::Write(const char* buffer, unsigned long len,
#ifdef SIMULATE #ifdef SIMULATE
///////////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////////////
// //
// NormSimObject Implementation // NormSimObject Implementation (dummy NORM_OBJECT_FILE or NORM_OBJECT_DATA)
// //
NormSimObject::NormSimObject(class NormSession* theSession, NormSimObject::NormSimObject(class NormSession* theSession,

45
common/normObject.h
View File

@ -70,7 +70,50 @@ class NormObject
bool IsPending(bool flush = true) const; bool IsPending(bool flush = true) const;
bool IsRepairPending() const; bool IsRepairPending() const;
bool IsPendingInfo() {return pending_info;} 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 // Methods available to server for transmission
bool NextServerMsg(NormObjectMsg* msg); bool NextServerMsg(NormObjectMsg* msg);

142
common/normSegment.cpp
View File

@ -18,12 +18,11 @@ bool NormSegmentPool::Init(unsigned int count, unsigned int size)
{ {
if (seg_list) Destroy(); if (seg_list) Destroy();
peak_usage = 0; peak_usage = 0;
overruns = 0; overruns = 0;
#ifdef SIMULATE #ifdef SIMULATE
// In simulations, don't really need big vectors for data // In simulations, don't really need big vectors for data
// since we don't actually read/write real data // since we don't actually read/write real data (for the most part)
size = MIN(size, (SIM_PAYLOAD_MAX+NormDataMsg::PayloadHeaderLength()+1)); size = MIN(size, (SIM_PAYLOAD_MAX+1));
#endif // SIMULATE #endif // SIMULATE
// This makes sure we get appropriate alignment // This makes sure we get appropriate alignment
unsigned int alloc_size = size / sizeof(char*); unsigned int alloc_size = size / sizeof(char*);
@ -184,14 +183,15 @@ bool NormBlock::IsRepairPending(UINT16 ndata, UINT16 nparity)
if (nparity) if (nparity)
{ {
UINT16 i = nparity; UINT16 i = nparity;
NormSegmentId nextId = (UINT16)pending_mask.FirstSet(); NormSegmentId nextId = 0;
GetFirstPending(nextId);
while (i--) while (i--)
{ {
// (TBD) for more NACK suppression, we could skip ahead // (TBD) for more NACK suppression, we could skip ahead
// if this bit is already set in repair_mask? // if this bit is already set in repair_mask?
repair_mask.Set(nextId); // set bit a parity can fill repair_mask.Set(nextId); // set bit a parity can fill
nextId++; nextId++;
nextId = (UINT16)pending_mask.NextSet(nextId); GetNextPending(nextId);
} }
} }
else if (size > ndata) else if (size > ndata)
@ -390,66 +390,68 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
NormRepairRequest req; NormRepairRequest req;
req.SetFlag(NormRepairRequest::SEGMENT); req.SetFlag(NormRepairRequest::SEGMENT);
if (repairInfo) req.SetFlag(NormRepairRequest::INFO); if (repairInfo) req.SetFlag(NormRepairRequest::INFO);
UINT16 nextId = (UINT16)repair_mask.FirstSet(); NormSymbolId nextId = 0;
UINT16 blockSize = size; if (GetFirstRepair(nextId))
NormRepairRequest::Form prevForm = NormRepairRequest::INVALID;
UINT16 segmentCount = 0;
UINT16 firstId = 0;
while (nextId < blockSize)
{ {
UINT16 currentId = nextId; UINT16 blockSize = size;
nextId = (UINT16)repair_mask.NextSet(++nextId); NormRepairRequest::Form prevForm = NormRepairRequest::INVALID;
if (!segmentCount) firstId = currentId; UINT16 segmentCount = 0;
UINT16 firstId = 0;
segmentCount++; while (nextId < blockSize)
// Check for break in consecutive series or end
if (((nextId - currentId) > 1) || (nextId >= blockSize))
{ {
NormRepairRequest::Form form; UINT16 currentId = nextId;
switch (segmentCount) 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: NormRepairRequest::Form form;
form = NormRepairRequest::INVALID; switch (segmentCount)
break; {
case 1: case 0:
case 2: form = NormRepairRequest::INVALID;
form = NormRepairRequest::ITEMS; break;
break; case 1:
default: case 2:
form = NormRepairRequest::RANGES; form = NormRepairRequest::ITEMS;
break; 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) } // end while (nextId < blockSize)
{ if (NormRepairRequest::INVALID != prevForm)
if (NormRepairRequest::INVALID != prevForm) cmd.PackRepairRequest(req); // (TBD) error check
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
return true; return true;
} // end NormBlock::AppendRepairAdv() } // end NormBlock::AppendRepairAdv()
@ -461,25 +463,25 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
bool pendingInfo, bool pendingInfo,
UINT16 segmentSize) UINT16 segmentSize)
{ {
ASSERT(pending_mask.FirstSet() < ndata); NormSegmentId nextId = 0;
NormSegmentId nextId, endId; NormSegmentId endId;
if (erasure_count > nparity) if (erasure_count > nparity)
{ {
// Request explicit repair // Request explicit repair
nextId = (UINT16)pending_mask.FirstSet(); GetFirstPending(nextId);
UINT16 i = nparity; UINT16 i = nparity;
// Skip nparity missing data segments // Skip nparity missing data segments
while (i--) while (i--)
{ {
nextId++; nextId++;
endId = (UINT16)pending_mask.NextSet(nextId); GetNextPending(nextId);
} }
endId = ndata + nparity; endId = ndata + nparity;
} }
else else
{ {
nextId = (UINT16)pending_mask.NextSet(ndata); nextId = ndata;
GetNextPending(nextId);
endId = ndata + erasure_count; endId = ndata + erasure_count;
} }
NormRepairRequest req; NormRepairRequest req;
@ -492,7 +494,7 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
while (nextId < endId) while (nextId < endId)
{ {
UINT16 currentId = nextId; UINT16 currentId = nextId;
nextId = (UINT16)pending_mask.NextSet(++nextId); if (!GetNextPending(++nextId)) nextId = endId;
if (0 == segmentCount) firstId = currentId; if (0 == segmentCount) firstId = currentId;
segmentCount++; segmentCount++;
// Check for break in consecutive series or end // Check for break in consecutive series or end
@ -796,7 +798,9 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock)
prev->next = theBlock; prev->next = theBlock;
else else
table[index] = theBlock; table[index] = theBlock;
ASSERT((entry ? (blockId != entry->Id()) : true)); ASSERT((entry ? (blockId != entry->Id()) : true));
theBlock->next = entry; theBlock->next = entry;
return true; return true;
} // end NormBlockBuffer::Insert() } // end NormBlockBuffer::Insert()

39
common/normSegment.h
View File

@ -156,10 +156,36 @@ class NormBlock
void IncrementParityCount() {parity_count++;} void IncrementParityCount() {parity_count++;}
UINT16 ParityCount() const {return parity_count;} UINT16 ParityCount() const {return parity_count;}
NormSymbolId FirstPending() const bool GetFirstPending(NormSymbolId& symbolId) const
{return (NormSymbolId)pending_mask.FirstSet();} {
NormSymbolId FirstRepair() const UINT32 index;
{return (NormSymbolId)repair_mask.FirstSet();} 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) bool SetPending(NormSymbolId s)
{return pending_mask.Set(s);} {return pending_mask.Set(s);}
bool SetPending(NormSymbolId firstId, UINT16 count) bool SetPending(NormSymbolId firstId, UINT16 count)
@ -190,11 +216,6 @@ class NormBlock
bool IsTransmitPending() const bool IsTransmitPending() const
{return (pending_mask.IsSet() || repair_mask.IsSet());} {return (pending_mask.IsSet() || repair_mask.IsSet());}
NormSymbolId NextPending(NormSymbolId index) const
{return ((NormSymbolId)pending_mask.NextSet(index));}
bool AppendRepairRequest(NormNackMsg& nack, bool AppendRepairRequest(NormNackMsg& nack,
UINT16 ndata, UINT16 ndata,
UINT16 nparity, UINT16 nparity,

66
common/normSession.cpp
View File

@ -170,13 +170,13 @@ bool NormSession::StartServer(unsigned long bufferSpace,
StopServer(); StopServer();
return false; 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"); DMSG(0, "NormSession::StartServer() tx_pending_mask.Init() error!\n");
StopServer(); StopServer();
return false; 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"); DMSG(0, "NormSession::StartServer() tx_repair_mask.Init() error!\n");
StopServer(); StopServer();
@ -278,9 +278,9 @@ void NormSession::Serve()
NormObject* obj = NULL; NormObject* obj = NULL;
// Queue next server message // 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); obj = tx_table.Find(objectId);
ASSERT(obj); ASSERT(obj);
} }
@ -937,12 +937,9 @@ void NormSession::HandleReceiveMessage(NormMsg& msg, bool wasUnicast)
if (IsClient()) ClientHandleNackMessage((NormNackMsg&)msg); if (IsClient()) ClientHandleNackMessage((NormNackMsg&)msg);
break; break;
case NormMsg::ACK: case NormMsg::ACK:
DMSG(4, "NormSession::HandleReceiveMessage(NormMsg::ACK) node>%lu ...\n",
LocalNodeId());
if (IsServer() && (((NormAckMsg&)msg).GetServerId() == LocalNodeId())) if (IsServer() && (((NormAckMsg&)msg).GetServerId() == LocalNodeId()))
ServerHandleAckMessage(currentTime, (NormAckMsg&)msg, wasUnicast); ServerHandleAckMessage(currentTime, (NormAckMsg&)msg, wasUnicast);
if (IsClient()) if (IsClient()) ClientHandleAckMessage((NormAckMsg&)msg);
ClientHandleAckMessage((NormAckMsg&)msg);
break; break;
case NormMsg::REPORT: case NormMsg::REPORT:
@ -1048,7 +1045,6 @@ double NormSession::CalculateRtt(const struct timeval& currentTime,
void NormSession::ServerUpdateGrttEstimate(double clientRtt) void NormSession::ServerUpdateGrttEstimate(double clientRtt)
{ {
grtt_response = true; grtt_response = true;
if (clientRtt > grtt_current_peak) if (clientRtt > grtt_current_peak)
{ {
@ -1361,23 +1357,19 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
NormObjectId txObjectIndex; NormObjectId txObjectIndex;
NormBlockId txBlockIndex; NormBlockId txBlockIndex;
if (tx_pending_mask.IsSet()) if (ServerGetFirstPending(txObjectIndex))
{ {
txObjectIndex = NormObjectId((unsigned short)tx_pending_mask.FirstSet());
NormObject* obj = tx_table.Find(txObjectIndex); NormObject* obj = tx_table.Find(txObjectIndex);
ASSERT(obj); ASSERT(obj && obj->IsPending());
if (obj->IsPending()) if (obj->IsPendingInfo())
{ {
if (obj->IsPendingInfo()) txBlockIndex = 0;
txBlockIndex = 0;
else
txBlockIndex = obj->FirstPending() + 1;
} }
else else
{ {
txObjectIndex = txObjectIndex + 1; obj->GetFirstPending(txBlockIndex);
txBlockIndex = 0; txBlockIndex++;
} }
} }
else else
{ {
@ -1447,8 +1439,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
ServerQueueSquelch(nextObjectId); ServerQueueSquelch(nextObjectId);
squelchQueued = true; squelchQueued = true;
} }
if ((OBJECT == requestLevel) || if ((OBJECT == requestLevel) || (INFO == requestLevel))
(INFO == requestLevel))
{ {
nextObjectId++; nextObjectId++;
if (nextObjectId > lastObjectId) inRange = false; if (nextObjectId > lastObjectId) inRange = false;
@ -1481,6 +1472,8 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
switch (requestLevel) switch (requestLevel)
{ {
case OBJECT: case OBJECT:
DMSG(8, "NormSession::ServerHandleNackMessage(OBJECT) objs>%hu:%hu\n",
(UINT16)nextObjectId, (UINT16)lastObjectId);
if (holdoff) if (holdoff)
{ {
if (nextObjectId > txObjectIndex) if (nextObjectId > txObjectIndex)
@ -1504,6 +1497,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
break; break;
case BLOCK: 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 // (TBD) if entire object is TxReset(), continue
if (object->IsStream()) if (object->IsStream())
{ {
@ -1577,6 +1573,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
inRange = false; inRange = false;
break; break;
case SEGMENT: 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 (nextBlockId != prevBlockId) freshBlock = true;
if (freshBlock) if (freshBlock)
{ {
@ -1635,7 +1634,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
if (1 == (txBlockIndex - nextBlockId)) if (1 == (txBlockIndex - nextBlockId))
{ {
// We're currently sending this block // We're currently sending this block
NormSegmentId firstPending = block->FirstPending(); ASSERT(block->IsPending());
NormSegmentId firstPending = 0;
block->GetFirstPending(firstPending);
if (lastLockId <= firstPending) if (lastLockId <= firstPending)
attemptLock = false; attemptLock = false;
else if (nextSegmentId < firstPending) else if (nextSegmentId < firstPending)
@ -1696,7 +1697,8 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else if (1 == (txBlockIndex - nextBlockId)) else if (1 == (txBlockIndex - nextBlockId))
{ {
NormSegmentId firstPending = block->FirstPending(); NormSegmentId firstPending = 0;
if (!block->GetFirstPending(firstPending)) ASSERT(0);
if (nextSegmentId > firstPending) if (nextSegmentId > firstPending)
object->TxUpdateBlock(block, nextSegmentId, lastSegmentId, numErasures); object->TxUpdateBlock(block, nextSegmentId, lastSegmentId, numErasures);
else if (lastSegmentId > firstPending) else if (lastSegmentId > firstPending)
@ -2223,8 +2225,12 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/)
} }
// 2) Update grtt_estimate _if_ sufficient time elapsed. // 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(); 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) if (grtt_response)
{ {
@ -2266,7 +2272,7 @@ bool NormSession::OnProbeTimeout(ProtoTimer& /*theTimer*/)
if (grtt_interval < grtt_interval_min) if (grtt_interval < grtt_interval_min)
grtt_interval = grtt_interval_min; grtt_interval = grtt_interval_min;
else else
grtt_interval *= 2.0; grtt_interval *= 1.5;
if (grtt_interval > grtt_interval_max) if (grtt_interval > grtt_interval_max)
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*)&currentTime.tv_sec); struct tm* ct = gmtime((time_t*)&currentTime.tv_sec);
DMSG(2, "Report time>%02d:%02d:%02d.%06lu node>%lu ***************************************\n", DMSG(2, "Report time>%02d:%02d:%02d.%06lu node>%lu ***************************************\n",
ct->tm_hour, ct->tm_min, ct->tm_sec, currentTime.tv_usec, LocalNodeId()); 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()) if (IsClient())
{ {
NormNodeTreeIterator iterator(server_tree); NormNodeTreeIterator iterator(server_tree);
NormServerNode* next; NormServerNode* next;
while ((next = (NormServerNode*)iterator.GetNextNode())) 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 rxRate = 8.0e-03*((double)next->RecvTotal()) / report_timer.GetInterval(); // kbps
double rxGoodput = 8.0e-03*((double)next->RecvGoodput()) / report_timer.GetInterval(); // kbps double rxGoodput = 8.0e-03*((double)next->RecvGoodput()) / report_timer.GetInterval(); // kbps
next->ResetRecvStats(); next->ResetRecvStats();

8
common/normSession.h
View File

@ -154,6 +154,14 @@ class NormSession
void ServerSetExtraParity(UINT16 extraParity) void ServerSetExtraParity(UINT16 extraParity)
{extra_parity = 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;} double ServerGroupSize() {return gsize_measured;}
void ServerSetGroupSize(double gsize) void ServerSetGroupSize(double gsize)
{ {

2
common/normVersion.h
View File

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

4
ns/simplenorm.tcl
View File

@ -121,7 +121,7 @@ $norm(1) repeat 50
$norm(1) interval 0.0 $norm(1) interval 0.0
$norm(1) txloss 10.0 $norm(1) txloss 10.0
$norm(1) gsize $groupSize $norm(1) gsize $groupSize
$norm(1) cc off $norm(1) cc on
#$norm(1) trace on #$norm(1) trace on
# 8) Configure NORM client agents at other nodes # 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($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" $ns_ at $duration "finish $ns_ $f $nf $nsTracing"