ProtocolStack/NACKManager.cs

using System;
using System.Collections;
using System.Threading;
using Common;

namespace ProtocolStack
{
        public class NACKManager
        {
                private NackQueue nackQueue; // queue of NACKs waiting to be sent out
                private TransmissionQueue txQueue; // the transmission queue in which to output things
                private RemoteHostComms thisRemoteHost;




                public NACKManager(RemoteHostComms thisRemoteHost) {
                        this.thisRemoteHost = thisRemoteHost;
                        nackQueue = new NackQueue();
                        this.txQueue = thisRemoteHost.TxQueue;
                }

                public void ResetConnections() {
                        Console.WriteLine("NACK Manager Reset Starting ...");
                        nackQueue.Clear();
                        Console.WriteLine("... NACK Manager Reset Complete");
                }

                public NackInfo GetNextNack() {
                        /* only one thread should be calling this at any given time, so no need to lock */
                        if (nackQueue.IsEmpty) { return null; }
                        else return nackQueue.Dequeue();
                }

                public void ProcessReceivedSegmentHeaders(ref SegmentHeaders newHeaders) {
                        /* examines headers to see which segments need to be retransmitted */
                        /* calculate end point of nack range 
                         * use it in a for loop to add a nack retransmision message to the relevant queue */
                        uint lastNACK;
                        uint firstNACK;
                        ushort destination;
                        //Console.WriteLine("NACK Manager: Process Received Segment Headers");
                        if (newHeaders.IsNACK) { // Retransmit
                                firstNACK = newHeaders.NackInfo.SequenceNumber;
                                if (newHeaders.NackInfo.IsRanged) { lastNACK = firstNACK + newHeaders.NackInfo.Range; }
                                else { lastNACK = firstNACK; }
                                destination = newHeaders.SourceDeviceID; // this should always be the same.

                                // TODO: Fix so that range can wrap round.
                                for (uint i = firstNACK; i <= lastNACK; i++) {
                                        // grab segment from retransmission buffer
                                        Segment tempSegment = thisRemoteHost.RetransmissionBuffer.GetSegment(i);
                                        // encapsulate segment in retransmission message
                                        RetransmissionMessage retransMessage = new RetransmissionMessage(tempSegment);
                                        // set the destination of the retransmission
                                        retransMessage.Destination = destination;
                                        // plonk message into the correct transmission queue
                                        TransmissionQueueEntry[] entries = TransmissionQueueEntry.FromMessage(retransMessage);
                                        txQueue.Enqueue(entries, false);
                                        Console.WriteLine("NACK Manager: Issued Reransmision of #" + i);
                                }
                        }
                }

                public void IssueNACK(uint segmentNumber) {
                        SendNACK(new NackInfo(segmentNumber));
                }

                public void IssueNACK(uint segmentNumber, long range) {
                        for (long offset = 0; offset < range; offset++) {
                                IssueNACK((uint)(segmentNumber + offset));
                        }
                        //SendNACK(new NackInfo(segmentNumber, (uint)range));
                }

                private void SendNACK(NackInfo nackInfo) {
                        lock (txQueue.SyncRoot) {
                                if (txQueue.IsEmpty) {
                                        /* create no-op message */
                                        Message noopMsg = new EmptyMessage();
                                        noopMsg.Destination = thisRemoteHost.RemoteDeviceID;
                                        /* segmentise */
                                        TransmissionQueueEntry[] buffer = TransmissionQueueEntry.FromMessage(noopMsg);
                                        /* place results in segment queue */
                                        // make sure this all happens in a monitor, so we get them all together
                                        txQueue.EnqueueUnprotected(buffer, false);
                                }
                                nackQueue.Enqueue(nackInfo);
                                Console.WriteLine("NACK Manager: NACK Sent for #" + nackInfo.SequenceNumber + " Range: " + (nackInfo.IsRanged ? nackInfo.Range.ToString() : "N/A"));
                                Monitor.PulseAll(txQueue.SyncRoot);
                        }
                }
        }

        /*****************************************************************************/
        public class NackInfo {
                bool isRanged;
                uint sequenceNumber;
                uint range;

                public NackInfo(uint sequenceNumber) {
                        this.isRanged = false;
                        this.sequenceNumber = sequenceNumber;
                }

                public NackInfo(uint sequenceNumber, uint range) {
                        this.isRanged = true;
                        this.sequenceNumber = sequenceNumber;
                        this.range = range;
                }

                public bool IsRanged { get { return isRanged; }}
                public uint SequenceNumber { get { return sequenceNumber; } }
                public uint Range { get { return range; } }


        }


        /*****************************************************************************/
        class NackQueue : Queue {
                public NackQueue() : base() {}

                public void Enqueue(NackInfo nackInfo) {
                        lock (this.SyncRoot) {
                                base.Enqueue(nackInfo);                 
                        } 
                }

                public new NackInfo Dequeue() { // this must not block and wait, because of where it's tested.
                        NackInfo result;
                        lock (this.SyncRoot) {
                                result = (NackInfo)base.Dequeue();
                        }
                        return result;
                }

                public bool IsEmpty {
                        get {
                                bool result;
                                lock (this.SyncRoot) {
                                        result = (this.Count == 0);
                                }
                                return result;
                        }
                }
        }

        /*****************************************************************************/
        public class RetransmissionBuffer {
                /* Contains a copy of every segment that may be requested by a 
                 * negative acknowledgement.
                 * 
                 * Data is indexed by segment number. Segment number and timestamp
                 * would have been better, but without adding another field to a
                 * segment we can not be sure of selecting the correct segment for
                 * retransmission.
                 * 
                 * If a segment is retransmitted, it will not need to be re-transmitted again
                 * (the segment encapsulating the retransmission will be retransmitted). So we 
                 * can erase the segment from the buffer when it is sent for retransmission.
                 * 
                 * If segment m is such that all segments < m have been received, and m is a NACK 
                 * requesting segment n, then we can retransmit 'n' and erase from the buffer
                 * all segments <= n. 
                 * 
                 * Let's just weaken this condition to 'if there are no missing segments'. 
                 * This is the case where m is the segment just received. And to be honest, 
                 * this is likely to be the case.
                 * 
                 * Failure case: some segment > m is carrying a NACK for segment < n. 
                 * This would occur if... there is no way this would occur.
                 * 
                 * Note that all segments < m must have been received, not NACKd.
                 * 
                 * As a failsafe we should have a 'Segment not in Buffer' message for problems.
                 * */
                Hashtable buffer;
                private SequenceManager seqManager;
                
                private uint earliestSegmentNumber {
                        get { 
                                // TODO: Correct Min for wrapround error
                                lock (buffer.SyncRoot) {
                                        ICollection keys = buffer.Keys;
                                        uint[] keyArray = new uint[keys.Count];
                                        keys.CopyTo(keyArray, 0);
                                        Array.Sort(keyArray);
                                        return keyArray[0];
                                }
                        }
                }

                public RetransmissionBuffer(SequenceManager seqManager) {
                        this.seqManager = seqManager;
                        buffer = new Hashtable();
                }




                public void ResetConnection() {
                        lock (buffer.SyncRoot) {
                                buffer.Clear();
                        }
                }


                private void Add(Segment target) {
                        lock (buffer.SyncRoot) {
                                buffer.Add(target.Headers.SequenceNumber, target);
                        }

                }

                private void Remove(uint segmentNumber) {
                        lock (buffer.SyncRoot) {
                                buffer.Remove(segmentNumber);
                        }
                }

                private Segment Read(uint segmentNumber) {
                        lock (buffer.SyncRoot) {
                                return (Segment)buffer[segmentNumber];
                        }
                }

                public Segment GetSegment(uint sequenceNumber) {
                        lock (buffer.SyncRoot) {
                                Segment result = null;

                                result = Read(sequenceNumber);
                                /* when we erase all segments 'less' than n, how do we identify them? They could wrap round,
                                 * and thus some could be > n. Perhaps we should have a linked list type index structure? */
                                /* or perhaps we should have some kind of 'earliest segment' indicator. */
                                if (!seqManager.IsSegmentsMissing) {
                                        // erase all segments less than nackSegmentHeaders.SequenceNumber
                                        // i.e. all the ones that were put in before it.
                                        if (earliestSegmentNumber > sequenceNumber) {
                                                for (long i = sequenceNumber; i >= 0; i--) {
                                                        this.Remove((uint)i); }
                                                for (long i = UInt32.MaxValue; i >= earliestSegmentNumber; i--) {
                                                        this.Remove((uint)i); }
                                        } else { // earliestSegmentNumber <= nackSegmentHeaders.SequenceNumber
                                                for (long i = sequenceNumber; i >= earliestSegmentNumber; i--) {
                                                        this.Remove((uint)i); }
                                        }
                                } // else  there are segments missing, so we can't erase anything for certain, so noop.
                                return result;
                        }
                }

                public void PutSegment(Segment target) {
                        // todo: handle exceptions, etc.
                        lock (buffer.SyncRoot) {
                                this.Add(target);
                        }
                }
        

        }
}

---