2
\$\begingroup\$

I'm writing a communications library for a class. Part of the requirements is that semi-reliable communications, which leads me to this monstrosity:

unit UDP.Communicator;

{$mode objfpc}{$H+}

interface

uses
  Classes, SysUtils, DateUtils, SyncObjs, blcksock, fgl, SimpleThread, Common.Types, UDP.Shared;

type

{ TUDPQueueCard }
TUDPQueueCard = class
public
  Guid: String;
  Address: TIPEndPoint;
  Packet: TUDPPacket;
  RetryAttempts: Integer;
  LastRetry: TDateTime;
  constructor Create;
  destructor Destroy; override;
end;
TUDPQueueCardMap = specialize TFPGMap<String, TUDPQueueCard>;

{ TUDPSeenMessageRecord }

TUDPSeenMessageRecord = class
public
  Guid: String;
  Seen: TDateTime;
end;
TUDPSeenMessageMap = specialize TFPGMap<String, TUDPSeenMessageRecord>;

TUDPCommunicator = Class(TObject)
private
  FSocket: TUDPBlockSocket;
  FSocketThread: TThread;
  FRunner: TUDPTaskRunner;
  FRecvQueue: TThreadList;
  FSendQueue: TThreadList;
  FSeenMessages: TUDPSeenMessageMap;
  FAwaitingAck: TUDPQueueCardMap;
  procedure Execute(Thread: TSimpleThread);
  procedure ProcessSendQueue;
  procedure ProcessRecvQueue;
  procedure ProcessTimeouts;
public
  OnDatagramReceived: TDatagramEvent;
  constructor Create;
  procedure Bind(AEndPoint: TIPEndPoint);
  procedure Listen;
  procedure Send(Envelope: TUDPEnvelope);
end;

implementation

function GetPacket(Envelope: TUDPEnvelope): TUDPPacket;
var
  Packet: TUDPPacket;
begin
  // Create stuff
  Packet := Default(TUDPPacket);
  Envelope.Stream.Position := 0;
  Packet.DataSize := Envelope.Stream.Size;
  Envelope.Stream.WriteBuffer(Packet.DataPacket, SizeOf(Packet.DataPacket));
  Packet.Guid := Envelope.GUID;
  Packet.Reliable := Envelope.Reliable;
  Packet.MessageType := 'PROCESS';
  Result := Packet;
end;


{ TUDPCommunicator }

constructor TUDPCommunicator.Create;
begin
  FSocket := TUDPBlockSocket.Create;

  FRecvQueue := TThreadList.Create;
  FSendQueue := TThreadList.Create;
  FAwaitingAck := TUDPQueueCardMap.Create;
  FSeenMessages := TUDPSeenMessageMap.Create;

  // Create the server
  FSocketThread := TSimpleThread.Create(@Self.Execute);

  // Create the task runner
  FRunner := TUDPTaskRunner.Create(false);

  FRunner.PQueue := @FRecvQueue;

  FRunner.FEventHandler := OnDatagramReceived;
end;

procedure TUDPCommunicator.Bind(AEndPoint: TIPEndPoint);
begin
  WriteLn('Attempting to Bind');
  if Assigned(FSocket) then
  begin
    WriteLn('Socket Assigned');
    FSocket.Bind(AEndPoint.Address, AEndPoint.Port);
    if FSocket.LastError <> 0 then
    begin
      WriteLn('Cannot bind: ' + FSocket.LastErrorDesc);
    end;
  end;
end;

procedure TUDPCommunicator.Listen;
begin
  if Assigned(FSocket) then
  begin
    FSocket.Listen;
  end;
end;

procedure TUDPCommunicator.Send(Envelope: TUDPEnvelope);
var
  SendLock: TList;
  QCard: TUDPQueueCard;
begin
  SendLock := FSendQueue.LockList;
  try
    // Turn the envelope into a queue card
    QCard := TUDPQueueCard.Create;
    QCard.Address.Address := Envelope.Address.Address;
    QCard.Address.Port := Envelope.Address.Port;
    QCard.Packet := GetPacket(Envelope);
    QCard.Guid := QCard.Packet.Guid;
    QCard.RetryAttempts := 0;
    QCard.LastRetry := Now;
    SendLock.Add(QCard);
  finally
    FSendQueue.UnlockList;
    FreeAndNil(Envelope);
  end;
end;

procedure TUDPCommunicator.Execute(Thread: TSimpleThread);
begin
  while not Thread.Terminated do
  begin
    ProcessSendQueue;
    ProcessRecvQueue;
    ProcessTimeouts;
      // Release the CPU
    Sleep(10);
  end;
end;

procedure TUDPCommunicator.ProcessSendQueue;
var
  Count: Integer;
  QCard: TUDPQueueCard;
  SendLock: TList;
begin
  if Assigned(FSendQueue) then
  begin
    SendLock := FSendQueue.LockList;
    try
      Count := SendLock.Count;
      if Count > 0 then
      begin
        // Get the first qcard from the queue
        QCard := TUDPQueueCard(SendLock.Items[0]);
        // Remove it from the queue
        SendLock.Delete(0);
        // Store data about the last attempt
        Inc(QCard.RetryAttempts);
        QCard.LastRetry := Now;

        // Attempt to send the packet
        try
          FSocket.Connect(QCard.Address.Address, QCard.Address.Port);
          FSocket.SendBuffer(@QCard.Packet, SizeOf(QCard.Packet));
          if FSocket.LastError = 0 then
          begin
            // Output to the User
            WriteLn('Sent: ' + QCard.Guid);

            // Make sure we're not adding ack packet after ack packet to
            // reception queues.
            if QCard.Packet.MessageType <> 'ACK' then
            begin
              //Add the message to the sent dictionary
              FAwaitingAck.Add(QCard.Guid, QCard);
              WriteLn('Added Message: ', QCard.Guid);
            end
          end
          else
          begin
            // Socket Error Require the Message
            WriteLn('Adding message due to socket error');
            FSendQueue.Add(QCard);
            WriteLn('Socket Error: ' + IntToStr(FSocket.LastError));
          end;
        except on E: Exception do
          WriteLn('unrecoverable error: ', E.ToString);
          //FreeAndNil(QCard);
        end;
      end;
    finally
      FSendQueue.UnlockList;
    end;
  end;
end;

procedure TUDPCommunicator.ProcessRecvQueue;
var
  ReadLock, WriteLock: TList;
  QCard, AckCard: TUDPQueueCard;
  SeenMsg: TUDPSeenMessageRecord;
begin
  if FSocket.CanRead(100) then
  begin
    WriteLn('Has Data in Buffer');

    // Create a new datagram packet
    // Get the packet string
    QCard := TUDPQueueCard.Create;
    FSocket.RecvBuffer(@QCard.Packet, SizeOf(QCard.Packet));
    // Why?  Why not?
    QCard.Guid := QCard.Packet.Guid;
    QCard.Address.Address := FSocket.GetRemoteSinIP;
    QCard.Address.Port := IntToStr(FSocket.GetRemoteSinPort);
    WriteLn('Message Received', QCard.Guid);

    case QCard.Packet.MessageType of

    'ACK':
      begin
        // remove the message from the AwaitingAck map
        AckCard := TUDPQueueCard(FAwaitingAck.KeyData[QCard.Guid]);
        FAwaitingAck.Remove(QCard.Guid);
        // Free the AckCard
        FreeAndNil(AckCard);
      end;

    'PROCESS':
      begin
        // Create an ack message
        AckCard := TUDPQueueCard.Create;
        AckCard.Address.Address := QCard.Address.Address;
        AckCard.Address.Port := QCard.Address.Port;

        // We only care about the message type & guid
        AckCard.Guid := QCard.Guid;
        AckCard.Packet.MessageType := 'ACK';
        AckCard.Packet.Guid := QCard.Guid;

        // Add the ack packet to the send queue and note that
        // we've seen the message
        WriteLock := FSendQueue.LockList;
        WriteLock.Add(AckCard);
        FSendQueue.UnlockList;

        // Check to see if this is a new message
        if FSeenMessages.IndexOf(QCard.Guid) < 0 then
        begin
          WriteLn('Message new to me! ', FSeenMessages.Count, ' messages previously seen');
          SeenMsg := TUDPSeenMessageRecord.Create;
          SeenMsg.Guid := AckCard.Guid;
          SeenMsg.Seen := Now;
          FSeenMessages.Add(SeenMsg.Guid, SeenMsg);
          if Assigned(FRecvQueue) then
          begin
            ReadLock := FRecvQueue.LockList;
            try
              ReadLock.Add(QCard);
            finally
              FRecvQueue.UnlockList;
            end; // try
          end; // if Assigned(PRecvQueue) then
        end; // if FSeenMessages.IndexOf(QCard.Guid) < 0 then
      end; // begin

    else
      begin
        WriteLn('Invalid message received');
      end;

    end; // case


  end; // if FSocket.CanRead(100) then
end;

procedure TUDPCommunicator.ProcessTimeouts;
var
  //QPair: T
  QCard: TUDPQueueCard;
  Lock: TList;
begin
  if FAwaitingAck.Count > 0 then
  begin
    QCard := TUDPQueueCard(FAwaitingAck.Data[0]);
    // Remove it so it will be sent to the back of the queue if it needs to be readded
    FAwaitingAck.Remove(QCard.Guid);
    if QCard.RetryAttempts > 3 then
    begin
      WriteLn('Timeout on Message: ', QCard.Guid);
    end
    else if SecondsBetween(Now, QCard.LastRetry) > 1 then
    begin
      // It has been >1 second since this message was sent and no response
      // was received.  Readd it to the send queue.
      WriteLn('Retrying Message', QCard.Guid);
      WriteLn('Last attempt ', MilliSecondsBetween(Now, QCard.LastRetry), 'ms ago.');
      Lock := FSendQueue.LockList;
      Lock.Add(QCard);
      FSendQueue.UnlockList;
    end else
      // No action needs to be taken at this time.  Just add it to the sent message queue
      // for later processing.
      FAwaitingAck.Add(QCard.Guid, QCard);
  end;
end;

constructor TUDPQueueCard.Create;
begin
  Address := TIPEndPoint.Create;
  Packet := Default(TUDPPacket);
end;

destructor TUDPQueueCard.Destroy;
begin
  FreeAndNil(Address);
  FreeAndNil(Packet);
end;

end.

What this does is attempts to send a message. If the message is received, the receiver sends an ACK message back to the sender. If no ack message is received, the message is resent 3 times until it times out. Since the ack message could be error prone, the system keeps track of messages it has seen. Obviously, there is A LOT to pick a part here. Big thank you in advance to anyone who can find a good way to sort this out a little bit.

One big problem I'm having is a massive amount of memory leaks. Does FreeAndNil(); do what I think it does? Which is to say, does it free the instance of an object? If so, why are my Envelopes not being freed?

\$\endgroup\$
  • 2
    \$\begingroup\$ So, basically, you're re-inventing TCP? \$\endgroup\$ – Mast Nov 30 '16 at 12:03
1
\$\begingroup\$

I would throw the code away and start over. I'd get exchanges working with small minimal test routines before evolving to classes/objects/libraries/etc.

You don't need anything fancy, like queues and such, for a synchronous system that doesn't send a new packet until the previous one is acked. The code below is essentially a client-server setup.

Included in each outgoing client UDP packet is an incrementing serial number. The send routine for the client should bump the current serial number, send the packet, then wait for the ack packet from the server, which is expected to include the same serial number.

The receive portion of code for the client (executed immediately after a send, or after you've done the send followed by some other work while the server's ack is on its way) waits for a packet or times out. If the incoming ack packet doesn't have the same serial number that was sent, keep reading (the correct ack packet might still be coming, or you'll eventually time out anyway). If you time out, and still want to retry, repeat the send using the same serial number and wait again for the ack.

function clientSentPacketAndGotAck: boolean; // not debugged
begin
result := FALSE; // assume failure
bumpSerialNumber;
for tries := 1 to 3 do
  begin
  sendDataPacket;
  repeat
    timedOut := FALSE;
    // can do other brief work here
    readAckPacketOrTimeOut; // sets timedOut if applicable
    result := (not timedOut) and sameSerialNumberReceived;
    if result then exit; // done, else timed out or got wrong serial number
  until timedOut;
  end; // timed out, do another send-receive
end; // 3 sends failed, result is FALSE

procedure server; // not debugged
begin
repeat // expecting serial number one greater than previous
  if somethingToRead then // O/S indicates incoming packet is ready to read
    begin // note that if continuous garbage arrives, no work gets done
    readPacket;
    if packetNotOurExpectedFormat then continue; // ignore junk
    sendAckPacketContainingSameSerialNumber;
    // might have just acked a repeated serial number we already processed
    // because client probably never received our previous ack
    if receivedSerialNumberIsOneGreaterThanPrevious then 
      begin
      processReceivedDataPacket;
      inc(expectedSerialNumber); // client and server must initially agree on this
      end;
    end
  else // do other work here
until shuttingDown;
end;

See RFC 1982 for the proper "greater than" algorithm when comparing serial numbers that can wrap around.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.