SegmentTransport.cs

using System;
using System.Net;
using System.Net.Sockets;
using System.Collections;
using System.Threading;

namespace ProtocolStack {

        public class SegmentTransmitter {
                RemoteHostComms thisRemoteHost; 
                StackInterface thisStack;
                UdpClient myUDP;

                Segment workingSegment;
                IPAddress currentDestination;
                Thread processingThread;

                public void Start() {
                        if (processingThread == null || !processingThread.IsAlive) {
                                Console.WriteLine("Starting Segment Transmitter...");
                                try {
                                        StackSettings s = thisStack.Settings;
                                        myUDP = new UdpClient(new IPEndPoint(thisStack.ipMappings.Lookup(thisStack.settings.MyDeviceID), s.MyDeviceID == s.SERVER_DEVICE_ID ? s.ServerSendFromPortNumber : s.ClientSendFromPortNumber));
                                } catch (ApplicationException appEx) {
                                        throw new ApplicationException(String.Format("Unable to create transmission socket", thisStack.ipMappings.Lookup(thisStack.settings.MyDeviceID)), appEx);
                                } catch (SocketException sockEx) {
                                        throw new ApplicationException(String.Format("Unable to create transmission socket for IP Address [{0}]", thisStack.ipMappings.Lookup(thisStack.settings.MyDeviceID)), sockEx);
                                }
                                processingThread = new Thread(new ThreadStart(Run));
                                processingThread.Start();
                                Console.WriteLine("...Segment Transmitter Started.");
                        }
                }

                public void Stop() {
                        if (processingThread != null && processingThread.IsAlive) {
                                processingThread.Abort();
                                processingThread.Join();
                                myUDP.Close();
                        }
                }

                public SegmentTransmitter(RemoteHostComms thisRemoteHost) {
                        this.thisRemoteHost = thisRemoteHost;
                        this.thisStack = thisRemoteHost.thisStack;
                }

                public void Run() {
                        /* Takes next segment from transmit queue. 
                         * Looks up destination IP
                         * Fills in any Just In Time fields
                         * Send (rate control?)
                         * Repeat As Needed */
                        try {
                                while (true) {
                                        // get next queue entry
                                        TransmissionQueueEntry currentQueueEntry; 
                                        if (thisRemoteHost.readyToSendUserData) {
                                                currentQueueEntry = thisRemoteHost.TxQueue.DequeueBlocking(); // blocks
                                        } else { // if we're waiting to be ready to send user daat
                                                Console.WriteLine("### Not ready to send ###");
                                                lock(thisRemoteHost.AckManager) {
                                                        Console.WriteLine("### Waiting for ACKable to be ready ###");
                                                        while (!thisRemoteHost.AckManager.IsAckableWaiting()) {
                                                                Monitor.Wait(thisRemoteHost.AckManager);
                                                                Console.WriteLine("Woke up");
                                                        }
                                                        Console.WriteLine("### ACKable ready, creating empty message carrier");
                                                        currentQueueEntry = TransmissionQueueEntry.FromMessage(new EmptyMessage(thisRemoteHost.RemoteDeviceID), true)[0];
                                                }
                                        }
                                        BlockForCreditAndSend(currentQueueEntry);
                                }
                        } catch (ThreadAbortException) {
                                // close down
                        }
                }

                private void BlockForCreditAndSend(TransmissionQueueEntry currentQueueEntry) {
                        bool originalUnsent = true;
                        while (originalUnsent) {
                                if (thisRemoteHost.RateManager.WaitForSend(currentQueueEntry.segmentData.DataLength, 10)) {  // if we're ok to send
                                        SendTQE(currentQueueEntry);
                                        originalUnsent = false;
                                } else {
                                        // construct no-op mesage and use that to carry credit info
                                        EmptyMessage msg = new EmptyMessage();
                                        msg.Destination = currentQueueEntry.destinationDeviceID;
                                        TransmissionQueueEntry[] entryArray = TransmissionQueueEntry.FromMessage(msg, true); // should only be 1 segment!
                                        Console.WriteLine("    10 seconds elapsed without credit to send - sending credit update message to try and free deadlock *");
                                        SendTQE(entryArray[0]);
                                        originalUnsent = true;
                                }
                        }
                }

                private void SendTQE(TransmissionQueueEntry currentQueueEntry) {
                        workingSegment = new Segment();
                        uint sequenceOffset = 0;
                        RateInfo rateInfo = null;
                        /* Gathering Data For Headers */
                        bool isAckable = thisRemoteHost.AckManager.IsAckableWaiting();
                        AckInfo ackInfo = thisRemoteHost.AckManager.GetNextAck();
                        workingSegment.Headers = new SegmentHeaders();
                        NackInfo nackInfo = thisRemoteHost.NackManager.GetNextNack();
                        uint seqNum = thisRemoteHost.SeqManager.GetNextSequenceNumber();
                        bool isResynch = thisRemoteHost.SeqManager.IsResynching;
                        if (isResynch) {
                                sequenceOffset = thisRemoteHost.SeqManager.GetSequenceOffset(seqNum);
                        } else {
                                rateInfo = thisRemoteHost.RateManager.GetRateInfoForTx(isAckable);
                        }
                        // TODO: Add Try / Catch here (looking up IP address of destination)
                        currentDestination = thisStack.ipMappings.Lookup(currentQueueEntry.destinationDeviceID); // unneeded? 

                        /* Assign Segment Data */
                        workingSegment.Data = currentQueueEntry.segmentData;

                        /* Fill in Segment Headers */
                        /* Initial Flags */
                        workingSegment.Headers.IsAckable = isAckable;
                        workingSegment.Headers.IsResynch = isResynch;
                        workingSegment.Headers.IsFirstInMessage = currentQueueEntry.isMessageStart;
                        workingSegment.Headers.IsLastInMessage = currentQueueEntry.isMessageEnd;

                        /* Source Device ID and Sequence Number */
                        workingSegment.Headers.SourceDeviceID = thisStack.settings.MyDeviceID;
                        workingSegment.Headers.SequenceNumber = seqNum;

                        /* ACKs and NACKs */
                        if (ackInfo == null) { // no ACKs today
                                workingSegment.Headers.ClearAck();
                        } else { // ACK that segment!
                                workingSegment.Headers.SetAck(ackInfo);
                        }
                        if (nackInfo == null) { // no NACKs today
                                workingSegment.Headers.ClearNACKStatus();
                        } else { // NACK that segment!
                                workingSegment.Headers.SetNACKStatus(nackInfo);
                                Console.WriteLine("Adding NACK on outgoing segment #{0} for retrans of seg #{1}", workingSegment.Headers.SequenceNumber, nackInfo.SequenceNumber);
                        }
                                
                        /* Synch or rate info */
                        if (isResynch) {
                                workingSegment.Headers.SynchOrRate = sequenceOffset;
                        } else {
                                workingSegment.Headers.SynchOrRate = rateInfo.data;
                        }

                        /* Time stamp and send */
                        workingSegment.Headers.TimeSent = DateTime.Now.ToUniversalTime();
                        tempSend(workingSegment, currentDestination);
                        thisRemoteHost.RetransmissionBuffer.PutSegment(workingSegment);
                        thisRemoteHost.RateManager.DecreaseSendCredit(workingSegment.Data.DataLength);

                        /* If we sent an Ackable packet, make a note to expect a reply */
                        if (isAckable) { 
                                thisRemoteHost.AckManager.SentAckableSegment(workingSegment.Headers.SequenceNumber);
                        }
                }

                public void tempSend(Segment targetSegment, IPAddress destination) {
                        byte[] datagram = targetSegment.ToByteArray();
                        StackSettings s = thisStack.Settings;
                        myUDP.Send(datagram, datagram.Length, new IPEndPoint(destination, (s.MyDeviceID == 0) ? s.ClientReceivePortNumber : s.ServerReceivePortNumber));
                        Console.WriteLine("  > Sent #{0}, Credit Onboard: {1}, ACKable: {2}, Size: {3}", targetSegment.Headers.SequenceNumber, targetSegment.Headers.SynchOrRate ,targetSegment.Headers.IsAckable, targetSegment.Data.DataLength);
                        //Console.WriteLine("  > Sent #{0}", targetSegment.Headers.SequenceNumber);
                }
        }

        /*************************************************************************/

        public class SegmentReceiver {
                /* Run's in it's own thread. 
                 * Gabs UDP packets from the UDP Rx port
                 * Turns those packets into segments and stamps a Rx time on them
                 * Sends segment to appropriate next stage (DeMux)*/
                Thread processingThread;
                StackInterface thisStack;
                UdpClient udpListener;

                public ReceiveQueue destinationQueue;

                public SegmentReceiver(StackInterface thisStack) {
                        this.thisStack = thisStack;
                }

                public void Start() {
                        Console.WriteLine("Starting Segment Receiver...");
                        if (processingThread == null || !processingThread.IsAlive) {
                                processingThread = new Thread(new ThreadStart(Listen));
                                processingThread.Start();
                                Console.WriteLine("... Segment Receiver Started.");
                        } else {
                                Console.WriteLine("... Segment Receiver Already Started!");
                        }
                        
                }

                public void Stop() {
                        Console.WriteLine("Stopping Segment Receiver...");
                        if (processingThread != null && processingThread.IsAlive) {
                                processingThread.Abort();
                                if (udpListener != null) { udpListener.Close(); }
                                processingThread.Join();
                                Console.WriteLine("... Segment Receiver Stopped.");
                        } else {
                                Console.WriteLine("... Segment Receiver Already Stopped!");
                        }
                }

                private void Listen() { 
                        IPEndPoint currentEndPoint;
                        Segment receivedSegment;
                        int localPortNumber = -1;
                        if (thisStack.settings.MyDeviceID == thisStack.Settings.SERVER_DEVICE_ID) { // we are the server
                                localPortNumber = thisStack.Settings.ServerReceivePortNumber;
                        } else {// we are a client
                                localPortNumber = thisStack.Settings.ClientReceivePortNumber;
                        }
                        // create UDP socket bound to listen on IP address from settings
                        udpListener = new UdpClient(new IPEndPoint(IPAddress.Any, localPortNumber));
                        try {
                                while(true) {/* We want to do as little as possible here so we can run this in a tight loop */
                                        try {
                                                // create end point to hold receieved address
                                                currentEndPoint = new IPEndPoint(IPAddress.Any, localPortNumber);
                                                receivedSegment = new Segment(udpListener.Receive(ref currentEndPoint));
                                                receivedSegment.Headers.TimeReceived = DateTime.Now.ToUniversalTime();
                                                Console.WriteLine("  < Got #{0}", receivedSegment.Headers.SequenceNumber);
                                                thisStack.GetRemoteHostByID(receivedSegment.Headers.SourceDeviceID).AcceptIncomingSegment(receivedSegment);
                                        } catch (ThreadAbortException) {
                                        } catch (Exception ex) {
                                                Console.Error.WriteLine("\n *** Error in Protocol Stack ***\nUnable to correctly receive segment.\nSkipping over and continuting. Details:\n\n(" + ex + ")");
                                        }
                                }
                        } catch (ThreadAbortException) {
                                Console.WriteLine("Segment Receive Listen Thread Aborted");
                        } finally {
                                if (udpListener != null) { udpListener.Close(); }
                        }
        }
}

        /*************************************************************************/
        public class ConnectionReset : ApplicationException {}
        /*************************************************************************/
        public delegate void ConnectionResetDelegate();
        /*************************************************************************/
}

---