UDP Receive Data

herbie497

Member
Joined
Jun 19, 2014
Messages
7
Programming Experience
3-5
I'm trying to write a UDP application that sends a message and waits for a response before sending another.
To do this I am using a Semaphore initialising with the total threads, up to 6, and a count of 1.

In the Semaphore thread, I have a loop that runs as long as the semaphore is not complete ie. doesn't have a response.

The issue is with UDP, I'm fairly new to this class, and have been struggling a lot.
So far I am successful with sending the data, but have experienced exceptions when trying to receive.
In the UDP Manager Class, I have created an Rx thread which listens, and adds the data to a buffer. The buffer gets called each time a Semaphore thread is waiting for a response.

Semaphore thread:
VB.NET:
  Private Sub Init_Semaphore(ByVal TotalThreads As Integer)
    ListBox1.Items.Add("== Start ======")
    Dim SemaphoreCount As Integer = 1
    Dim Threads(TotalThreads) As Thread
    Dim Sema As Semaphore = New Semaphore(SemaphoreCount, SemaphoreCount)

    TotalCompleted = 0

    For i As Integer = 0 To TotalThreads - 1
      UDPManager.AddCommandToBuffer(IPAddress, TxPort, RxPort, MessageList.Item(i), "J" & MessageList.Item(i).Substring(0, 1))
      Threads(i) = New Thread(New ParameterizedThreadStart(AddressOf AccessCode_WithSemaphore))
      Threads(i).IsBackground = True
      Threads(i).Name = MessageList.Item(i)
      Threads(i).Start(Sema)
    Next
  End Sub


  Private Sub AccessCode_WithSemaphore(ByVal ObjSemaphore As Object)
    Dim IsComplete As Boolean = False
    Dim l_Semaphore As Semaphore = DirectCast(ObjSemaphore, Semaphore)

    Do While IsComplete = False
      Dim Message As String = Thread.CurrentThread.Name
      Dim Response As UDP_Manager.Cls_UDPManager.Struct_Response = Nothing
      If l_Semaphore.WaitOne(200, False) Then
        Try
          ListBox1.BeginInvoke(New ParameterizedThreadStart(AddressOf UpdateUILB1), New Object() {"Thread ID: J" & Message.Substring(0, 1) & " Sending message: " & Message})

          Do Until Response.Response <> Nothing
            UDPManager.ReceiveMessage("J" & Message.Substring(0, 1))
            Response = UDPManager.GetMessageFromRxBuffer("J" & Message.Substring(0, 1))
          Loop

          Thread.Sleep(50)
        Catch ex As Exception

        Finally
          l_Semaphore.Release()
          ListBox1.BeginInvoke(New ParameterizedThreadStart(AddressOf UpdateUILB1), New Object() {"Thread ID: J" & Message.Substring(0, 1) & " Message received: " & Response.Response})
          IsComplete = True
          'SemaphoreReady = True
          TotalCompleted += 1

          If TotalCompleted = MessageList.Count Then
            SemaphoreReady = True
          End If
        End Try
      Else
        'Waiting to enter
      End If
    Loop
  End Sub

Rx Thread:
VB.NET:
  Private Sub ReceiveCallback(ByVal ar As IAsyncResult)
    Dim u As UdpClient = CType((CType(ar.AsyncState, Cls_UDPState)).UDPClient, UdpClient)
    Dim e As IPEndPoint = CType((CType(ar.AsyncState, Cls_UDPState)).EndPoint, IPEndPoint)

    Try
      Dim ReceiveBytes() As Byte = u.EndReceive(ar, e)
      Dim ReceiveString As String = Encoding.ASCII.GetString(ReceiveBytes)
      If Not ReceiveString = Nothing Then
        RxBufferInUse = True
        Dim Response As New Struct_Response
        With Response
          .Response = ReceiveString.Substring(0, ReceiveString.Length - 2)
          .Sender = CommandBuffer.Item(0).Sender
          .RxTime = Now
        End With
        ResponseBuffer.Add(Response)

        Threading.Thread.Sleep(300)

        If CommandBufferInUse = False Then
          CommandBufferInUse = True
          CommandBuffer.RemoveAt(0)
          CommandBufferInUse = False
        End If

      Else
        Dim i As Integer = 0
      End If
    Catch ex As Exception

    End Try

    RxBufferInUse = False
    SendNextMessage = True
  End Sub

  Public Sub ReceiveMessage(ByVal Sender As String)
    If CommandBuffer.Count > 0 Then
      Dim EndPoint As IPEndPoint
      EndPoint = New IPEndPoint(IPAddress.Any, 0)

      If RxUDP Is Nothing Then
        Try
          RxUDP = New UdpClient(CommandBuffer.Item(0).IPConfig.RxPort)
          State = New Cls_UDPState
          State.EndPoint = EndPoint
          State.UDPClient = RxUDP
          RxThread = New AsyncCallback(AddressOf ReceiveCallback)
        Catch ex As Exception

        End Try
      End If

      Try
        RxUDP.BeginReceive(RxThread, State)
      Catch ex As Exception
        [COLOR=#ff0000]'Mostly fails here with this exception:
        'System.Net.Sockets.SocketException = {"An operation on a socket could not be performed because the system lacked sufficient buffer space or because a queue was full"}[/COLOR]
      End Try
    End If
  End Sub

  Public Function GetMessageFromRxBuffer(ByVal Sender As String) As Struct_Response
    RxBufferInUse = True
    If ResponseBuffer.Count > 0 Then
      For i As Integer = ResponseBuffer.Count - 1 To 0 Step -1
        If ResponseBuffer.Count > 0 Then
          Try
            If ResponseBuffer.Item(i).Sender = Sender Then
              Dim Response As Struct_Response = ResponseBuffer.Item(i)
              If ResponseBuffer.Count > 0 Then
                Try
                  ResponseBuffer.RemoveAt(i)
                Catch ex As Exception
                  Dim j As Integer = 0
                End Try
                RxBufferInUse = False
                Return Response
              Else
                Exit For
              End If 'response count
            End If 'sender
          Catch ex As Exception
            Dim j As Integer = 0
          End Try
        Else
          Exit For
        End If 'response count
      Next
    Else

    End If 'response count

    RxBufferInUse = False
    Return Nothing

  End Function
 
I still have an issue with this, and haven't got any further. I have tried to increase the RxUDP.Client.Buffersize, and I'm only sending one message at a time so can't see it being the queue is full.
 
In your ReceiveCallback you need to Close the UdpClient.

You should also look at this part "Do Until Response.Response <> Nothing" where each loop calls ReceiveMessage, that may create new UdpClient. There is probably a better way to synchronize than a polling loop, perhaps using events or a WaitHandle.
 
Thanks for you reply.

I've changed the receive process to
VB.NET:
  Private Sub ReceiveCallback(ByVal ar As IAsyncResult)
    Dim u As UdpClient = CType((CType(ar.AsyncState, Cls_UDPState)).UDPClient, UdpClient)
    Dim e As IPEndPoint = CType((CType(ar.AsyncState, Cls_UDPState)).EndPoint, IPEndPoint)

    Try
      ReceiveBytes = u.EndReceive(ar, e)

      TestString = TestString & System.Text.Encoding.ASCII.GetString(ReceiveBytes.ToArray)

      'Dim ReceiveString As String = Encoding.ASCII.GetString(ReceiveBytes)
      Dim ReceiveString As String = TestString
      If Not ReceiveString = Nothing Then

        ReceiveString = ReceiveString.Replace(vbCr, "")
        ReceiveString = ReceiveString.Replace(vbLf, "")

        Dim testarray() As String = ReceiveString.Split(",")

        Dim ItemIndex As Integer
        For Each Item As Struct_CommandInfo In TxBuffer
          If Item.Command.Contains(ReceiveString.ToLower) Then
            ItemIndex = Item.Command.IndexOf(ReceiveString.ToLower)
            Exit For
          End If
        Next

        Dim commandsentarray() As String = TxBuffer.Item(ItemIndex).Command.Split(",")

        If testarray(0) = commandsentarray(0) Then
          If testarray(1) = commandsentarray(1).Replace("?", "").ToUpper Then
            For i As Integer = 2 To testarray.Count - 2
              If IsNumeric(testarray(i)) = False Then
                Console.WriteLine(Now & " Fail! Receive values are not valid. Retrying.")
                WaitTimer.Stop()
                WaitTimer.Reset()
                SendNextMessage = True
                Exit Sub
              End If
            Next
          Else
            Console.WriteLine(Now & " Fail! Receive command does not match sent command. Retrying. " & testarray(1))
            WaitTimer.Stop()
            WaitTimer.Reset()
            SendNextMessage = True
            Exit Sub
          End If
        Else
          Console.WriteLine(Now & " Fail! Receive address does not match sent address. Retrying. " & testarray(0))
          WaitTimer.Stop()
          WaitTimer.Reset()
          SendNextMessage = True
          Exit Sub
        End If

        RxBufferInUse = True
        Dim Response As New Struct_Response
        With Response
          .Response = ReceiveString
          .Sender = TxBuffer.Item(ItemIndex).Sender
          .TxTime = TxBuffer.Item(ItemIndex).TxTime
          .RxTime = Now
        End With

        If Response.Response.Contains("*") Then
          RxBuffer.Add(Response)
          TestString = ""
          WaitTimer.Stop()
          WaitTimer.Reset()
        End If

        RxBufferInUse = False
        SendNextMessage = True

        Do
          If TxBufferInUse = False Then
            TxBufferInUse = True
            If Response.Sender = TxBuffer.Item(ItemIndex).Sender Then
              TxBuffer.RemoveAt(ItemIndex)
            End If
            TxBufferInUse = False
            Exit Do
          End If
        Loop

      Else
        Dim i As Integer = 0
      End If
    Catch ex As Exception
      Console.WriteLine(ex.Message)
      Dim i As Integer = 0
    End Try

    u.Close()
    RxUDP.Close()
  End Sub


  Public Sub ReceiveMessage(ByVal Sender As String)
    If TxBuffer.Count > 0 Then
      Dim EndPoint As IPEndPoint
      'EndPoint = New IPEndPoint(IPAddress.Any, TxBuffer.Item(0).IPConfig.RxPort)
      EndPoint = New IPEndPoint(IPAddress.Any, 0)

      If RxUDP Is Nothing Then
        Try
          RxUDP = New UdpClient
          RxUDP.ExclusiveAddressUse = False
          RxUDP.Client.SetSocketOption(Net.Sockets.SocketOptionLevel.Socket, Net.Sockets.SocketOptionName.ReuseAddress, True)
          RxUDP.Client.Bind(EndPoint)

          State = New Cls_UDPState
          State.EndPoint = EndPoint
          State.UDPClient = RxUDP

          RxThread = New AsyncCallback(AddressOf ReceiveCallback)
        Catch ex As Exception
          Console.WriteLine(ex.Message)
          Dim i As Integer = 0
          Exit Sub
        End Try
      End If

      Try
        RxUDP.BeginReceive(RxThread, State)
      Catch ex As Exception
        Console.WriteLine("RxUDP.BeginReceive(): " & ex.Message)
        Dim j As Integer = 0
      End Try
    End If
  End Sub

With the Semaphore loop now as this:
VB.NET:
  Private Sub AccessCode_WithSemaphore(ByVal ObjSemaphore As Object)
    Dim IsComplete As Boolean = False
    Dim l_Semaphore As Semaphore = DirectCast(ObjSemaphore, Semaphore)
    Dim TimeOut As Integer = 5000
    Dim TimeoutTimer As New Stopwatch

    TimeoutTimer.Start()
    Do While IsComplete = False
      Dim Message As String = Thread.CurrentThread.Name
      Dim Response As UDP_Manager.Cls_UDPManager.Struct_Response = Nothing

      If l_Semaphore.WaitOne(500, False) Then
        Try
          ListBox.BeginInvoke(New ParameterizedThreadStart(AddressOf UpdateUILB1), New Object() {"Thread ID: J" & Message.Substring(0, 1) & " Sending message: " & Message})

          Dim ItemIndex As Integer
          For Each Item As UDP_Manager.Cls_UDPManager.Struct_CommandInfo In UDPManager.TxBuffer
            If Item.Command.Contains(Message) Then
              ItemIndex = Item.Command.IndexOf(Message)
              Exit For
            End If
          Next

          Call UDPManager.SendCommandToClient(ItemIndex)
          Call UDPManager.ReceiveMessage("J" & Message.Substring(0, 1))

          Do Until Response.Response <> Nothing
            Response = UDPManager.GetMessageFromRxBuffer("J" & Message.Substring(0, 1))

            If Response.Response = Nothing And TimeoutTimer.ElapsedMilliseconds > TimeOut Then
              Response.Response = "Timed Out"
              Exit Do
            End If

            Thread.Sleep(100)
          Loop

          TimeoutTimer.Stop()
          TimeoutTimer.Reset()

        Catch ex As Exception
          Console.WriteLine(ex.Message)
          Dim i As Integer = 0
        Finally
          l_Semaphore.Release()
          ListBox1.BeginInvoke(New ParameterizedThreadStart(AddressOf UpdateUILB1), New Object() {"Thread ID: J" & Message.Substring(0, 1) & " Message received: '" & Response.Response & "'"})
          IsComplete = True
          TotalCompleted += 1

          If TotalCompleted = MessageList.Count Then
            SemaphoreReady = True
          End If
        End Try
      Else
        'Waiting to enter
      End If
    Loop
  End Sub

I don't get any exceptions but also don't seem to get a response what-so-ever; I have placed a breakpoint in ReceiveCallback() and this never gets hit. I know the message is being sent and I know the hardware responds because I have checked using a separate terminal window. Unfortunately I cannot use this terminal to test my application because replicating how the hardware responds is practically impossible.
 
This may have no impact overall, but in ReceiveCallback you do RxUDP.Close.
Issue #1 is that RxUDP is not a local or passed object to ReceiveCallback, so ReceiveCallback should not touch it.
Issue #2, isn't RxUDP actually the same object as the passed "u" UdpClient object? Or is it? If you use multiple threads accessing a global member will get you into trouble, avoid that and implement OOP principles. Keep things local and pass things around with parameters, don't assign them globally and assume they will be relevant in any thread or time context.

I can't follow all your code flow, but AccessCode_WithSemaphore somewhere calls SendCommandToClient and ReceiveMessage. ReceiveMessage may create UdpClient, then calls BeginReceive pointing to ReceiveCallback method. In ReceiveMessage you are again accessing global objects, there should be no reason for this. For example, if you only have one "RxUDP" what happens if you call BeginReceive again when another thread has already such a call waiting to receive?

As a single thread operation you have way to much "garbage" in code (a single async read operation is not that complicated), and as a multithread operation and don't see how it can work at all.
 
I wasn't sure which client I needed to close, so I did both. I'm guessing it should only be "u.Close()"?

My code flow should be:
Add commands to buffer
Start threads
Send thread command
Wait for response (for 5 secs)
Print response
- or if no response, remove command from buffer
Allow next thread to send command

Starting with AccessCode_WithSemaphore() and ending with ReceiveCallback() for each command.


ReceiveMessage should only create a single UDPClient due to:
If RxUDP Is Nothing Then
 Try
          RxUDP = New UdpClient
          RxUDP.ExclusiveAddressUse = False
          RxUDP.Client.SetSocketOption(Net.Sockets.SocketOptionLevel.Socket, Net.Sockets.SocketOptionName.ReuseAddress, True)
          RxUDP.Client.Bind(EndPoint)

          State = New Cls_UDPState
          State.EndPoint = EndPoint
          State.UDPClient = RxUDP

          RxThread = New AsyncCallback(AddressOf ReceiveCallback)
        Catch ex As Exception
          Console.WriteLine(ex.Message)
          Dim i As Integer = 0
          Exit Sub
        End Try
      End If


I have just attempted to raise an event for when a response is received.

In ReceiveCallback()
If Response.Response.Contains("*") Then
  RxBuffer.Add(Response)
  TestString = ""
  WaitTimer.Stop()
  WaitTimer.Reset()
  RaiseEvent MessageReceived(Response)
End If


Semaphore loop
Call UDPManager.ReceiveMessage("J" & Message.Substring(0, 1))
Do Until Response.Response <> Nothing
  'Response = UDPManager.GetMessageFromRxBuffer("J" & Message.Substring(0, 1))

  If Response.Response = Nothing And TimeoutTimer.ElapsedMilliseconds > TimeOut Then
    ResponseReceived = "Timed Out"

    Do
      If UDPManager.TxBufferInUse = False Then
        UDPManager.TxBufferInUse = True
        UDPManager.TxBuffer.RemoveAt(ItemIndex)
        UDPManager.TxBufferInUse = False
        Exit Do
      End If
    Loop
    Exit Do
  End If

Loop



Event handler on the form
Private Sub Form_Load(sender As Object, e As EventArgs) Handles MyBase.Load
  AddHandler UDPManager.MessageReceived, AddressOf MessageReceived
End Sub

Private ResponseReceived As String
Private Sub MessageReceived(ByVal Response As Cls_UDPManager.Struct_Response)
  ResponseReceived = UDPManager.GetMessageFromRxBuffer("J" & Response.Response.Substring(0, 1)).Response
End Sub


I have no way of testing this as now I am getting this exception "ex.Message = Cannot access a disposed object. Object name: 'System.Net.Sockets.UdpClient'."
On line: "ReceiveCallback(): ReceiveBytes = u.EndReceive(ar, e)"
 
ReceiveMessage should only create a single UDPClient due to:
If RxUDP Is Nothing Then
I noticed that, and that was also partly what I commented about.
If your design is that you should only have one UDPClient then you also only need one concurrent communication thread, because another thread can't actually do any work.
If your design is to have one UDPClient for each of multiple threads, then make sure that each thread does not step on each others toes. This is done by either passing any state along the method calls and not touching anything outside, or using class instances to keep state separate for each instance. "Each to their own" is the motto.
I have no way of testing this as now I am getting this exception "ex.Message = Cannot access a disposed object. Object name: 'System.Net.Sockets.UdpClient'."
On line: "ReceiveCallback(): ReceiveBytes = u.EndReceive(ar, e)"
Oh-oh, one thread has stepped on another ones toes.
 
So, you're saying I need a client for each thread? So I'll need a list of clients with their indexes synchronised with the command message list?
Or a single client that never closes the connection?
 
I'm trying to write a UDP application that sends a message and waits for a response before sending another.
For this you only need one UDPClient. Of course, if you are sending/receiving from different remote endpoints concurrently you need one for each.

This also imply that you need a queue of messages to send, and not send next until first is processed, which in turn would require a time check (use a Timer, not a StopWatch) if no response is received (server may send it, but client may not receive it). UDPClient that is then waiting with BeginReceive will have to be closed for it to terminate the missing receive. Sending next can be done at BeginReceive callback regardless. Beware though, if you use a timeout and that message arrive later, then your communication protocol will be out of sync, since your next read will actually read the late previous message.

A simpler approach if server (and communication protocol) allows it is to just send any message when needed, and have the UDPClient listen for any response that may arrive. UDPClient will allow you to send while same instance at the same time is waiting with a BeginReceive, this can receive (and wait to receive) for application lifetime. This is simpler because there will be no management for UDP communications, all management will take place at the list of messages received (and possibly synced with the list of messages that was sent).
now I am getting this exception "ex.Message = Cannot access a disposed object. Object name: 'System.Net.Sockets.UdpClient'."
This was actually due to closing the UDPClient, but RxUDP is still referencing the now disposed object, so check "If RxUDP Is Nothing Then" is not True. A problem in design is in place here.
Do Until Response.Response <> Nothing
Still a bad idea, you don't need this because you know when a message has been received at BeginReceive callback. You can from there raise the event about message received, and if needed start next send.
 
Back
Top