{

for (unsigned i = 0; i < SiameseEncoderStats_Count; ++i) {

Counts[i] = 0;

}

{

NextColumn = 0;

ColumnStart = 0;

ClearWindow();

{

FirstUnremovedElement = 0;

Count = 0;

LongestPacket = 0;

SumStartElement = 0;

SumEndElement = 0;

for (unsigned laneIndex = 0; laneIndex < kColumnLaneCount; ++laneIndex)

{

EncoderColumnLane& lane = Lanes[laneIndex];

for (unsigned sumIndex = 0; sumIndex < kColumnSumCount; ++sumIndex)

{

lane.Sum[sumIndex].Bytes = 0;

lane.NextElement[sumIndex] = laneIndex;

}

lane.LongestPacket = 0;

}

{

if (EmergencyDisabled) {

return Siamese_Disabled;

}

if (GetRemainingSlots() <= 0) {

// This is not invalid input - The application often does not have

// control over how much data it tries to send. Instead it must

// listen for our feedback return code to decide when to slow down.

return Siamese_MaxPacketsReached;

}

const unsigned column = NextColumn;

const unsigned subwindowCount = Subwindows.GetSize();

unsigned element = Count;

// Assign packet number

packet.PacketNum = column;

// If there is not enough room for this new element:

// Note: Adding a buffer of kColumnLaneCount to create space ahead for

// snapshots as a subwindow is filled and we need to store its snapshot

if (element + kColumnLaneCount >= subwindowCount * kSubwindowSize)

{

EncoderSubwindow* subwindow = TheAllocator->Construct<EncoderSubwindow>();

if (!subwindow || !Subwindows.Append(subwindow))

{

EmergencyDisabled = true;

Logger.Error("WindowAdd.Construct OOM");

SIAMESE_DEBUG_BREAK();

return Siamese_Disabled;

}

}

if (Count > 0) {

++Count;

}

else

{

// Start a new window:

element = column % kColumnLaneCount;

StartNewWindow(column);

}

// Initialize original packet with received data

OriginalPacket* original = GetWindowElement(element);

if (0 == original->Initialize(TheAllocator, packet))

{

EmergencyDisabled = true;

Logger.Error("WindowAdd.Initialize OOM");

SIAMESE_DEBUG_BREAK();

return Siamese_Disabled;

}

SIAMESE_DEBUG_ASSERT(original->Column % kColumnLaneCount == element % kColumnLaneCount);

// Attach original send time in milliseconds

*GetWindowElementTimestampPtr(element) = static_cast<uint32_t>(GetTimeMsec());

// Roll next column to assign

NextColumn = IncrementColumn1(NextColumn);

// Update longest packet

const unsigned originalBytes = original->Buffer.Bytes;

const unsigned laneIndex = column % kColumnLaneCount;

EncoderColumnLane& lane = Lanes[laneIndex];

if (lane.LongestPacket < originalBytes) {

lane.LongestPacket = originalBytes;

}

if (LongestPacket < originalBytes) {

LongestPacket = originalBytes;

}

Stats->Counts[SiameseEncoderStats_OriginalCount]++;

Stats->Counts[SiameseEncoderStats_OriginalBytes] += packet.DataBytes;

return Siamese_Success;

{

// Maintain the invariant that element % 8 == column % 8 by skipping some

const unsigned element = column % kColumnLaneCount;

ColumnStart = column - element;

SIAMESE_DEBUG_ASSERT(column >= element && ColumnStart < kColumnPeriod);

SumStartElement = element;

SumEndElement = element;

FirstUnremovedElement = element;

Count = element + 1;

// Reset longest packet

LongestPacket = 0;

for (unsigned laneIndex = 0; laneIndex < kColumnLaneCount; ++laneIndex) {

Lanes[laneIndex].LongestPacket = 0;

}

Logger.Info(">>> Starting a new window from column ", ColumnStart);

{

if (EmergencyDisabled) {

return;

}

// Convert column to element, handling wrap-around:

const unsigned firstKeptElement = ColumnToElement(firstKeptColumn);

// If the column is outside of the window:

if (InvalidElement(firstKeptElement))

{

// If the element was before the window:

if (IsColumnDeltaNegative(firstKeptElement)) {

Logger.Info("Remove before column ", firstKeptColumn, " - Ignored before window");

}

else

{

// Removed everything

Count = 0;

Logger.Info("Remove before column ", firstKeptColumn, " - Removed everything");

}

}

else

{

Logger.Info("Remove before column ", firstKeptColumn, " element ", firstKeptElement);

// Mark these elements for removal next time we generate output

if (FirstUnremovedElement < firstKeptElement) {

FirstUnremovedElement = firstKeptElement;

}

}

{

// Recreate all the sums from scratch after this:

for (unsigned laneIndex = 0; laneIndex < kColumnLaneCount; ++laneIndex)

{

// Calculate first element to accumulate for this lane

const unsigned nextElement = GetNextLaneElement(elementStart, laneIndex);

for (unsigned sumIndex = 0; sumIndex < kColumnSumCount; ++sumIndex)

{

Lanes[laneIndex].NextElement[sumIndex] = nextElement;

Lanes[laneIndex].Sum[sumIndex].Bytes = 0;

}

}

SumStartElement = elementStart;

SumEndElement = elementStart;

SumColumnStart = ElementToColumn(elementStart);

SumErasedCount = 0;

{

const unsigned firstKeptSubwindow = FirstUnremovedElement / kSubwindowSize;

const unsigned removedElementCount = firstKeptSubwindow * kSubwindowSize;

SIAMESE_DEBUG_ASSERT(firstKeptSubwindow >= 1);

SIAMESE_DEBUG_ASSERT(firstKeptSubwindow < Subwindows.GetSize());

SIAMESE_DEBUG_ASSERT(removedElementCount % kColumnLaneCount == 0);

SIAMESE_DEBUG_ASSERT(removedElementCount <= FirstUnremovedElement);

Logger.Info("******** Removing up to ", FirstUnremovedElement, " and startColumn=", ColumnStart);

// If there are running sums:

if (SumEndElement > SumStartElement)

{

// Roll up the sums past the removal point

for (unsigned laneIndex = 0; laneIndex < kColumnLaneCount; ++laneIndex)

{

for (unsigned sumIndex = 0; sumIndex < kColumnSumCount; ++sumIndex)

{

GetSum(laneIndex, sumIndex, removedElementCount);

SIAMESE_DEBUG_ASSERT(Lanes[laneIndex].NextElement[sumIndex] >= removedElementCount);

Lanes[laneIndex].NextElement[sumIndex] -= removedElementCount;

}

}

if (removedElementCount > SumStartElement) {

SumErasedCount += removedElementCount - SumStartElement;

}

if (SumEndElement > removedElementCount) {

SumEndElement -= removedElementCount;

}

else {

SumEndElement = 0;

}

if (SumStartElement > removedElementCount) {

SumStartElement -= removedElementCount;

}

else {

SumStartElement = 0;

}

}

// Shift kept subwindows to the front of the vector:

// Resize a temporary buffer for removed subwindows

if (!SubwindowsShift.SetSize_NoCopy(firstKeptSubwindow))

{

SIAMESE_DEBUG_BREAK(); // OOM

EmergencyDisabled = true;

return;

}

// Store removed subwindows temporarily

memcpy(

SubwindowsShift.GetPtr(0),

Subwindows.GetPtr(0),

firstKeptSubwindow * sizeof(EncoderSubwindow*)

);

const unsigned subwindowsShifted = Subwindows.GetSize() - firstKeptSubwindow;

// Shift subwindows to front that are being kept

memmove(

Subwindows.GetPtr(0),

Subwindows.GetPtr(firstKeptSubwindow),

subwindowsShifted * sizeof(EncoderSubwindow*)

);

// Removed subwindows are moved to the end (unordered) for later reuse

memcpy(

Subwindows.GetPtr(subwindowsShifted),

SubwindowsShift.GetPtr(0),

firstKeptSubwindow * sizeof(EncoderSubwindow*)

);

// Update the count of elements in the window

SIAMESE_DEBUG_ASSERT(Count >= removedElementCount);

Count -= removedElementCount;

// Roll up the ColumnStart member

ColumnStart = ElementToColumn(removedElementCount);

SIAMESE_DEBUG_ASSERT(ColumnStart == Subwindows.GetRef(0)->Originals[0].Column);

// Roll up the FirstUnremovedElement member

SIAMESE_DEBUG_ASSERT(FirstUnremovedElement % kSubwindowSize == FirstUnremovedElement - removedElementCount);

SIAMESE_DEBUG_ASSERT(FirstUnremovedElement >= removedElementCount);

FirstUnremovedElement -= removedElementCount;

// Determine the new longest packets

unsigned longestPacket = 0;

unsigned laneLongest[kColumnLaneCount] = { 0 };

for (unsigned i = FirstUnremovedElement, count = Count; i < count; ++i)

{

OriginalPacket* original = GetWindowElement(i);

const unsigned originalBytes = original->Buffer.Bytes;

if (longestPacket < originalBytes) {

longestPacket = originalBytes;

}

SIAMESE_DEBUG_ASSERT(original->Column % kColumnLaneCount == i % kColumnLaneCount);

const unsigned laneIndex = i % kColumnLaneCount;

if (laneLongest[laneIndex] < originalBytes) {

laneLongest[laneIndex] = originalBytes;

}

}

// Update longest packet fields

LongestPacket = longestPacket;

for (unsigned laneIndex = 0; laneIndex < kColumnLaneCount; ++laneIndex) {

Lanes[laneIndex].LongestPacket = laneLongest[laneIndex];

}

// If there are no running sums:

if (SumEndElement <= SumStartElement) {

ResetSums(FirstUnremovedElement);

}

{

unsigned nextColumnExpected = 0;

int headerBytes = DeserializeHeader_PacketNum(data, bytes, nextColumnExpected);

if (headerBytes < 1) {

SIAMESE_DEBUG_BREAK(); // Invalid input

return false;

}

data += headerBytes, bytes -= headerBytes;

// If we received ack data out of order:

unsigned columnDelta = SubtractColumns(nextColumnExpected, NextColumnExpected);

if (IsColumnDeltaNegative(columnDelta)) {

// Ignore any older data since it would lose information.

return true;

}

/*

if (NextColumnExpected == nextColumnExpected &&

Data && bytes == DataBytes &&

0 == memcmp(data, Data, bytes))

{

return true;

}

// Update next expected column

NextColumnExpected = nextColumnExpected;

// Reset message decoder state

Offset = 0;

LossColumn = NextColumnExpected;

LossCount = 0;

DataBytes = bytes;

// If there are NACK ranges:

if (bytes > 0)

{

// Copy the new data into place with some padding at the end

Data = TheAllocator->Reallocate(Data, bytes + kPaddingBytes,

pktalloc::Realloc::Uninitialized);

memcpy(Data, data, bytes);

memset(Data + bytes, 0, kPaddingBytes); // Zero guard bytes

// Returns false if decoding the first loss range fails

if (!DecodeNextRange()) {

return false;

}

SIAMESE_DEBUG_ASSERT(IsIteratorAtFront()); // Invalid data

}

// Update RTO estimation based on the latest loss ranges and rollup

UpdateRTO();

// Remove data the given column

TheWindow->RemoveBefore(NextColumnExpected);

return true;

{

const unsigned windowCount = TheWindow->Count;

// If there is no unacknowledged data:

unsigned firstLoss = TheWindow->ColumnToElement(NextColumnExpected);

if (firstLoss >= windowCount) {

return;

}

// Get current time

const uint64_t nowMsec64 = GetTimeMsec();

const uint32_t nowMsec = static_cast<uint32_t>(nowMsec64);

unsigned longestAckDelayMsec = 0;

// Locate first unacknowledged element

unsigned element = TheWindow->ColumnToElement(NextRTOColumn);

if (element >= windowCount)

{

Logger.Warning("NextRTOColumn was invalid, so rolling up to FirstUnremovedElement = ", TheWindow->FirstUnremovedElement);

element = TheWindow->FirstUnremovedElement;

}

// First walk through all the rollups:

for (; element < firstLoss; ++element)

{

// Get last time this was sent (maybe the second time)

const uint32_t* sendMsecPtr = TheWindow->GetWindowElementTimestampPtr(element);

SIAMESE_DEBUG_ASSERT(*sendMsecPtr != 0);

// Calculate delay since it was sent

const int32_t delayMsec = nowMsec - *sendMsecPtr;

if ((unsigned)delayMsec > longestAckDelayMsec && delayMsec > 0)

{

longestAckDelayMsec = delayMsec;

Logger.Info("ACKED TS = ", *sendMsecPtr, " for ID=", element + TheWindow->ColumnStart, " <- new max delay: ", delayMsec);

}

else

{

Logger.Info("acked ts = ", *sendMsecPtr, " for ID=", element + TheWindow->ColumnStart, " - not max delay - ", delayMsec);

}

}

// Scan the selective acknowledgements implied by the NACK ranges:

unsigned remaining = DataBytes;

const uint8_t* data = Data;

while (remaining > 0)

{

// Decode NACK loss range

unsigned relativeStart, lossCountM1;

const int lossRangeBytes = DeserializeHeader_NACKLossRange(

data, remaining + kPaddingBytes, relativeStart, lossCountM1);

if (lossRangeBytes < 0 ||

lossRangeBytes > (int)remaining)

{

SIAMESE_DEBUG_BREAK(); // Invalid input

return;

}

data += lossRangeBytes, remaining -= lossRangeBytes;

// If element has not rolled up to the first acknowledged packet:

if (element + 1 < firstLoss) {

element = firstLoss - 1; // Roll it up

}

// Calculat the first loss in this NACK range

firstLoss += relativeStart;

// Note that element may already be beyond this NACK range because we

// may have seen this NACK range in a prior Acknowledgement message.

// For each element up to the first loss:

for (; element < firstLoss; ++element)

{

if (element >= windowCount) {

SIAMESE_DEBUG_BREAK(); // Invalid input

return;

}

// Get last time this was sent (maybe the second time)

const uint32_t* sendMsecPtr = TheWindow->GetWindowElementTimestampPtr(element);

SIAMESE_DEBUG_ASSERT(*sendMsecPtr != 0);

// Calculate delay since it was sent

const int32_t delayMsec = nowMsec - *sendMsecPtr;

if ((unsigned)delayMsec > longestAckDelayMsec && delayMsec > 0)

{

longestAckDelayMsec = delayMsec;

Logger.Info("NACKED TS = ", *sendMsecPtr, " for ID=", element + TheWindow->ColumnStart, " <- new max delay: ", delayMsec);

}

else

{

Logger.Info("nacked ts = ", *sendMsecPtr, " for ID=", element + TheWindow->ColumnStart, " - not max delay - ", delayMsec);

}

}

// Update loss offset to keep in sync with NACK range encoder

firstLoss += lossCountM1 + 2;

}

// Update next expected column for RTT calculation

NextRTOColumn = TheWindow->ElementToColumn(element);

// If there was no delay measurement:

if (longestAckDelayMsec <= 0) {

return;

}

// Recalculate the window size based on current timeout

static const uint64_t kMaxWindowMsec = 4000;

static const uint64_t kMinWindowMsec = 100;

uint64_t windowMsec = RetransmitTimeoutMsec * 2;

if (windowMsec < kMinWindowMsec) {

windowMsec = kMinWindowMsec;

}

else if (windowMsec > kMaxWindowMsec) {

windowMsec = kMaxWindowMsec;

}

// Update estimate of maximum RTT

MaxWindowedRTT.Update(

longestAckDelayMsec,

nowMsec64,

windowMsec);

const unsigned RTT_max = MaxWindowedRTT.GetBest();

// Pick an RTO that is 50% higher than the windowed maximum

RetransmitTimeoutMsec = (RTT_max * 3) / 2;

// Do not allow the RTO to get so small that OS scheduling delays might cause timeouts.

static const unsigned kMinimumRTOMsec = 20;

if (RetransmitTimeoutMsec < kMinimumRTOMsec) {

RetransmitTimeoutMsec = kMinimumRTOMsec;

}

Logger.Info("Calculated windowMsec = ", windowMsec, " and RTT_max = ", RTT_max, " and RetransmitTimeoutMsec = ", RetransmitTimeoutMsec, " from longestAckDelayMsec = ", longestAckDelayMsec, " NextRTOColumn=", NextRTOColumn);

{

// If there is no more loss range data to process:

if (Offset >= DataBytes) {

return false;

}

// Decode loss range format:

SIAMESE_DEBUG_ASSERT(Data && DataBytes >= 1);

unsigned relativeStart, lossCountM1;

const int lossRangeBytes = DeserializeHeader_NACKLossRange(

Data + Offset, DataBytes + kPaddingBytes - Offset, relativeStart, lossCountM1);

if (lossRangeBytes < 0) {

return false;

}

Offset += lossRangeBytes;

if (Offset > DataBytes) {

SIAMESE_DEBUG_BREAK(); // Invalid input

// TBD: Disable codec here?

return false;

}

// Move ahead the loss column

LossColumn = AddColumns(LossColumn, relativeStart);

LossCount = lossCountM1 + 1;

return true;

{

if (LossCount <= 0)

{

// Note: LossColumn is used as the offset for the next loss range, so

// we should increment it to one beyond the end of the current region

// when we get to the end of the region.

LossColumn = IncrementColumn1(LossColumn);

if (!DecodeNextRange()) {

return false;

}

}

columnOut = LossColumn;

LossColumn = IncrementColumn1(LossColumn);

--LossCount;

return true;

{

// Reset message decoder state

Offset = 0;

LossColumn = NextColumnExpected;

LossCount = 0;

DecodeNextRange();

// Note: Ignore return value

{

// Reset message decoder state

Offset = 0;

LossColumn = 0;

LossCount = 0;

DataBytes = 0;

// Reset the oldest column

FoundOldest = false;

{

Window.TheAllocator = &TheAllocator;

Window.Stats = &Stats;

Ack.TheAllocator = &TheAllocator;

Ack.TheWindow = &Window;

const uint8_t* data,

unsigned bytes,

unsigned& nextExpectedPacketNumOut)

{

if (Window.EmergencyDisabled) {

return Siamese_Disabled;

}

if (!Ack.OnAcknowledgementData(data, bytes)) {

return Siamese_InvalidInput;

}

// Report the next expected packet number

nextExpectedPacketNumOut = Ack.NextColumnExpected;

Stats.Counts[SiameseEncoderStats_AckCount]++;

Stats.Counts[SiameseEncoderStats_AckBytes] += bytes;

return Siamese_Success;

{

const unsigned headerBytes = original->HeaderBytes;

SIAMESE_DEBUG_ASSERT(headerBytes > 0 && original->Buffer.Bytes > headerBytes);

const unsigned length = original->Buffer.Bytes - headerBytes;

#ifdef SIAMESE_DEBUG

// Check: Deserialize length from the front

unsigned lengthCheck;

int headerBytesCheck = DeserializeHeader_PacketLength(original->Buffer.Data, original->Buffer.Bytes, lengthCheck);

if (lengthCheck != length || (int)headerBytes != headerBytesCheck ||

headerBytesCheck < 1 || lengthCheck == 0 ||

lengthCheck + headerBytesCheck != original->Buffer.Bytes)

{

SIAMESE_DEBUG_BREAK(); // Invalid input

return 0;

}

#endif // SIAMESE_DEBUG

originalOut.PacketNum = original->Column;

originalOut.Data = original->Buffer.Data + headerBytes;

originalOut.DataBytes = length;

return length;

OriginalPacket* original,

SiameseOriginalPacket& originalOut)

{

// Load original data and length

const unsigned length = LoadOriginal(original, originalOut);

if (length == 0) {

Window.EmergencyDisabled = true;

return Siamese_Disabled;

}

Stats.Counts[SiameseEncoderStats_RetransmitCount]++;

Stats.Counts[SiameseEncoderStats_RetransmitBytes] += length;

return Siamese_Success;

{

originalOut.Data = nullptr;

originalOut.DataBytes = 0;

if (Window.EmergencyDisabled) {

return Siamese_Disabled;

}

// If there is no unacknowledged data:

if (Window.GetUnacknowledgedCount() <= 0)

{

Ack.FoundOldest = false; // Reset Ack state

return Siamese_NeedMoreData;

}

// Get the window element first the first column expected by the receiver

const unsigned firstElement = SubtractColumns(Ack.NextColumnExpected, Window.ColumnStart);

const unsigned count = Window.Count;

// If the peer's next expected column is outside the window:

if (IsColumnDeltaNegative(firstElement) ||

firstElement < Window.FirstUnremovedElement ||

firstElement >= count)

{

return Siamese_NeedMoreData;

}

const uint32_t nowMsec = static_cast<uint32_t>(siamese::GetTimeMsec());

const uint32_t retransmitMsec = Ack.RetransmitTimeoutMsec;

// If the oldest column is already found:

if (Ack.FoundOldest)

{

// Calculate the oldest element

const unsigned element = SubtractColumns(Ack.OldestColumn, Window.ColumnStart);

// If the oldest element is still within the window:

if (!IsColumnDeltaNegative(element) &&

element >= firstElement &&

element < count)

{

OriginalPacket* original = Window.GetWindowElement(element);

SIAMESE_DEBUG_ASSERT(original->Buffer.Data != nullptr);

SIAMESE_DEBUG_ASSERT(original->Buffer.Bytes > 0);

uint32_t* lastSendMsecPtr = Window.GetWindowElementTimestampPtr(element);

const uint32_t lastSendMsec = *lastSendMsecPtr;

SIAMESE_DEBUG_ASSERT(lastSendMsec != 0);

// If RTO has not elapsed for the oldest entry:

if ((uint32_t)(nowMsec - lastSendMsec) < retransmitMsec) {

// Keep FoundOldest true for next time and stop for now

return Siamese_NeedMoreData;

}

// Update last send time

*lastSendMsecPtr = nowMsec;

// Look again for the oldest column next time

Ack.FoundOldest = false;

Logger.Debug("Retransmitting oldest in window. RTO = ", retransmitMsec, ", ID=", original->Column);

return AttemptRetransmit(original, originalOut);

}

// Keep searching for the oldest column

Ack.FoundOldest = false;

}

// Find the first column ready to go, or the oldest column in the set:

// If the first one is already ready to go:

unsigned nackElement = firstElement;

OriginalPacket* oldestOriginal = Window.GetWindowElement(nackElement);

uint32_t* firstSendMsecPtr = Window.GetWindowElementTimestampPtr(nackElement);

uint32_t oldestSendMsec = *firstSendMsecPtr;

// If the first packet RTO expired:

if ((uint32_t)(nowMsec - oldestSendMsec) >= retransmitMsec)

{

// Update last send time

*firstSendMsecPtr = nowMsec;

Logger.Debug("Retransmitting first in window. RTO = ", retransmitMsec, ", ID=", oldestOriginal->Column);

return AttemptRetransmit(oldestOriginal, originalOut);

}

// If NACK list is available:

if (Ack.HasNegativeAcknowledgements())

{

// Restart loss iterator for next time

Ack.RestartLossIterator();

// While there is another NACK column to process:

unsigned column;

while (Ack.GetNextLossColumn(column))

{

// If the receiver has already seen all the packets we sent:

nackElement = Window.ColumnToElement(column);

if (nackElement >= count) {

break; // Stop here

}

// Lookup original packet and send time

OriginalPacket* original = Window.GetWindowElement(nackElement);

SIAMESE_DEBUG_ASSERT(original->Buffer.Data != nullptr);

SIAMESE_DEBUG_ASSERT(original->Buffer.Bytes > 0);

uint32_t* lastSendMsecPtr = Window.GetWindowElementTimestampPtr(nackElement);

const uint32_t lastSendMsec = *lastSendMsecPtr;

SIAMESE_DEBUG_ASSERT(lastSendMsec != 0);

// If RTO expired:

if ((uint32_t)(nowMsec - lastSendMsec) >= retransmitMsec)

{

// Update last send time

*lastSendMsecPtr = nowMsec;

Logger.Debug("Retransmitting NACK: RTO = ", retransmitMsec, ", ID=", original->Column);

return AttemptRetransmit(original, originalOut);

}