3
\$\begingroup\$

I'm relatively new to .Net/UWP networking and have been trying to make sense of the APIs. I'd like to implement a TCP client that is capable of reading and sending messages independently (there is no protocol and all messages are independent).

Typically (e.g. in a POSIX C++ server), this would be done by spinning up a thread and reading continuously (sleeping periodically if no data) or using asynchronous I/O (epoll, select-poll, etc.)

I've implemented a UWP-compatible client that attempts to use ReadAsync as well as cancellation tokens to shut the client down.

This is a pretty rough cut and I'd like to flesh it out more, especially wrt error handling. Some questions:

  1. Is looping continuously as I've done in ReadMessages() the recommended/idiomatic way of using these functions?

  2. Am I using the cancellation tokens appropriately? My intent is to be able to kill a ReadAsync() request when the client object is destructed (or, eventually, when a Disconnect() method is added).

  3. Where should I be detecting disconnects? try-catch blocks around the ReadAsync methods?

Lastly, messages (their byte[] buffers, rather), would be output using a fast queue. I've commented that portion out. The intent is to use this as a component in a larger application (e.g., Unity).

namespace Messages
{
  [StructLayout(LayoutKind.Sequential, Pack = 1)]
  public struct Header
  {
    public UInt32 size;
    public UInt32 id;
    public UInt64 received_timestamp;
  }

  [StructLayout(LayoutKind.Sequential, Pack = 1)]
  public struct PingMessage
  {
    public Header header;
    public UInt64 timestamp;

    public void Init()
    {
      header.size = (UInt32)Marshal.SizeOf(typeof(PingMessage));
      header.id = 0xdeaddead;
      timestamp = (UInt64)System.Diagnostics.Stopwatch.GetTimestamp();
    }
  }

  [StructLayout(LayoutKind.Sequential, Pack = 1)]
  public struct PongMessage
  {
    public Header header;
    public UInt64 ping_timestamp;

    public void Init()
    {
      header.size = (UInt32)Marshal.SizeOf(typeof(PongMessage));
      header.id = 0xdeadbeef;
    }
  }
}
  public class Client
  {
    private Windows.Networking.Sockets.StreamSocket m_socket;
    private System.Threading.Tasks.Task m_connectionTask;
    private Windows.Storage.Streams.DataWriter m_writer;
    private System.IO.Stream m_inputStream;

    private Windows.Networking.HostName m_hostname;
    private string m_port;

    private bool m_connected = false;
    private System.Threading.CancellationTokenSource m_cancellationSource = new System.Threading.CancellationTokenSource();

    private byte[] m_header = new byte[Marshal.SizeOf(typeof(Messages.Header))];
    private byte[] m_messageBuffer = null;

    //private BlockingRingBuffer<byte[]> m_receiveQueue = new BlockingRingBuffer<byte[]>(1024 * 1024, true);

    public static byte[] Serialize<T>(ref T str)
    {
      int size = Marshal.SizeOf(str);
      byte[] arr = new byte[size];
      GCHandle h = default(GCHandle);
      try
      {
        h = GCHandle.Alloc(arr, GCHandleType.Pinned);
        Marshal.StructureToPtr<T>(str, h.AddrOfPinnedObject(), false);
      }
      finally
      {
        if (h.IsAllocated)
        {
          h.Free();
        }
      }
      return arr;
    }

    public static T Deserialize<T>(byte[] arr) where T : struct
    {
      T str = default(T);
      GCHandle h = default(GCHandle);
      try
      {
        h = GCHandle.Alloc(arr, GCHandleType.Pinned);
        str = (T)Marshal.PtrToStructure(h.AddrOfPinnedObject(), typeof(T));
      }
      finally
      {
        if (h.IsAllocated)
        {
          h.Free();
        }
      }
      return str;
    }

    public async void Send(byte[] bytes)
    {
      await m_connectionTask;

      if (!m_connected)
      {
        Console.WriteLine("NOT CONNECTED: dropping message");
        return;
      }

      try
      {
        m_writer.WriteBytes(bytes);
        await m_writer.StoreAsync();
        Console.WriteLine("Sent " + bytes.Length + " bytes");
      }
      catch (Exception exception)
      {
        // If this is an unknown status it means that the error if fatal and retry will likely fail.
        if (Windows.Networking.Sockets.SocketError.GetStatus(exception.HResult) == Windows.Networking.Sockets.SocketErrorStatus.Unknown)
        {
          throw;
        }
        Console.WriteLine("Error: Failed to send: " + exception.Message);
      }
    }

    /*
    public byte[] GetNextMessage()
    {
      byte[] buffer;
      m_receiveQueue.TryDequeue(out buffer);
      return buffer;
    }
    */

    private async void ReadMessages()
    {
      //TODO: handle disconnections and exceptions?
      while (true)
      {
        // Read header
        await m_inputStream.ReadAsync(m_header, 0, m_header.Length, m_cancellationSource.Token);
        if (m_cancellationSource.IsCancellationRequested)
        {
          return;
        }

        // Read remainder of message
        Messages.Header header = Deserialize<Messages.Header>(m_header);
        m_messageBuffer = new byte[header.size];
        Buffer.BlockCopy(m_header, 0, m_messageBuffer, 0, m_header.Length);
        await m_inputStream.ReadAsync(m_messageBuffer, m_header.Length, (int)header.size - m_header.Length, m_cancellationSource.Token);
        if (m_cancellationSource.IsCancellationRequested)
        {
          return;
        }

        // Enqueue
        //TODO: put on ring buffer
        Console.WriteLine("Read message: {0:x}, {1} bytes, thread={2}", header.id, header.size, System.Threading.Thread.CurrentThread.ManagedThreadId);
        m_messageBuffer = null;
      }
    }

    public async void Connect(bool blockUntilConnected, Action<bool, Exception> OnConnected = null)
    {
      m_connectionTask = System.Threading.Tasks.Task.Run(
        async () =>
        {
          Exception e = null;

          try
          {
            await m_socket.ConnectAsync(m_hostname, m_port);
            m_writer = new Windows.Storage.Streams.DataWriter(m_socket.OutputStream);
            m_inputStream = m_socket.InputStream.AsStreamForRead();
            m_connected = true;
            Console.WriteLine("Remote Address: " + m_socket.Information.RemoteAddress);
          }
          catch (Exception exception)
          {
            Console.WriteLine("Error: Failed to connect: " + exception.Message);
            m_connected = false;
            e = exception;
          }

          if (OnConnected != null)
          {
            OnConnected(m_connected, e);
          }
        }
      );

      if (blockUntilConnected)
      {
        m_connectionTask.Wait();
      }

      await m_connectionTask;

      if (m_connected)
      {
        ReadMessages();
      }
    }

    public Client(string hostname, int port)
    {
      m_socket = new Windows.Networking.Sockets.StreamSocket();
      m_hostname = new Windows.Networking.HostName(hostname);
      m_port = port.ToString();
    }

    ~Client()
    {
      m_cancellationSource.Cancel();
      //m_receiveQueue.StopWaiting(); // queue cannot be reused after this
      m_writer.Dispose();
      m_socket.Dispose();
    }
  }
\$\endgroup\$

migrated from stackoverflow.com Feb 14 at 4:47

This question came from our site for professional and enthusiast programmers.

  • \$\begingroup\$ This is too broad for stackoverflow. You might want to take this to Code Review instead \$\endgroup\$ – TheGeneral Feb 14 at 4:37
  • \$\begingroup\$ ReadAsync: "Returns Task<Int32> A task that represents the asynchronous read operation. The value of the TResult parameter contains the total number of bytes read into the buffer. The result value can be less than the number of bytes requested" \$\endgroup\$ – Damien_The_Unbeliever Feb 14 at 7:15
  • \$\begingroup\$ Good catch, @Damien_The_Unbeliever! Seems I can't edit my post, though, but I've fixed that by checking the task result and repeatedly looping and awaiting. \$\endgroup\$ – trzy Feb 14 at 8:00
  • \$\begingroup\$ I had a quick look. To understand what you are doing there should be more comments. Some of the comments you have are just commented out code, which is usually consider poor form. The Connect method doesn't seem to be called anywhere, so I presume it's the main entry point. I think you need a comment stating what it is meant to do. Some explanation of what is going on in Serialize probably wouldn't go amiss either. I suggest thinking more about a reader who has little or no idea about what you are doing, in writing clear explanations you may help yourself understand the problem better. HTH. \$\endgroup\$ – George Barwood Feb 14 at 23:44
  • \$\begingroup\$ Thanks, George. This is just a client module. Connect() would be called by whoever is using the Client class. Serialization/Deserialization just converts message structures to byte[] arrays and back for sending and receiving. I've since removed these altogether (messages now serialize themselves). The key thing I was hoping to get feedback on is whether the async ReadMessages() makes sense. Once connected, I spin this up and it continuously listens for messages (using background worker threads provided by the OS). The user of client can call GetNextMessage() or Send() independently. \$\endgroup\$ – trzy Feb 15 at 8:04

Your Answer

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

Browse other questions tagged or ask your own question.