Asynchronous hardware communication - Library Design

Question

I've got some equipment ("tool") that communicates over CANbus that I need to communicate with. I'm using a USB->CAN dongle that provides a basic .Net control library. I'm in the process of writing a library that abstracts out tool communication, and I'd like some comments on the design.

Note that I've mostly been living in a .NET 2.0 world for a very, very long time. I've done some crash course style study of some of the new threading concepts in .NET 4.5, but I'm sure my code is a mix of old and new style.

Here is a brief overview of the architecture:

• The CAN library (PCANbasic) allows one to scan for a usb dongle and open a connection. One can write CAN packets to the connection, or poll for received packets.
• The tool has a custom communication scheme where a command or response can consist of one or more CAN packets.
• The goal of my library is to hide the PCAN library. I want to be able to open a connection (CanbusConnection) and be able to send commands and receive command responses (each command is an instance of an abstract class, CommandBase).
• All the communication has to happen on a background thread so it doesn't stall the UI.

CanbusConnection.cs

This class just wraps the underlying PCAN library to establish a connection, then it starts a background thread that polls for incoming CAN messages every ~100ms. It exposes a CanbusPacketReceived event. Closing the CanbusConnection class stops the background thread, then closes the underlying PCAN connection.

public delegate void CanbusPacketReceivedEventHandler(object sender, TPCANMsg packet);
public class CanbusConnection : IDisposable
{
    #region Declarations
    TPCANHandle _handle;
    Thread _canbusListenerThread;
    #endregion

    #region Properties
    public bool IsOpen { get; private set; }
    public event CanbusPacketReceivedEventHandler CanbusPacketReceived;
    #endregion

    public CanbusConnection()
    {
        _handle = 0;
        IsOpen = false;
    }

    public bool Open()
    {
        if (IsOpen) { return true; }

        try
        {
            IsOpen = TryInitializePCANChannel(out _handle);
            if(IsOpen)
            {
                _canbusListenerThread = new Thread(new ThreadStart(CanbusListener));
                _canbusListenerThread.Start();
            }
        }
        catch(Exception ex)
        {
            Log.Error("Error opening PCAN connection: {Message}, Error: {Error}", ex.Message, ex);
            throw ex;
        }
        return success;

    }

    private void CanbusListener()
    {
        TPCANMsg msg;

        while (IsOpen)
        {
            try
            {
                while(PCANBasic.Read(_handle, out msg) == TPCANStatus.PCAN_ERROR_OK)
                { 
                    string hexData = string.Join(" ", msg.DATA.Select(b => b.ToString("X2")));
                    Log.Information("EventType: {EventType}, Data: {Data}", CanbusConnectionEventType.CAN_RX, hexData);
                    NotifyCanbusPacketReceived(msg);
                }
            }
            catch (Exception ex)
            {
                Log.Warning("EventType: {EventType}, Message: {Message}", CanbusConnectionEventType.CAN_RX, ex.Message);
            }
            Thread.Sleep(100);
        }
    }

    private void NotifyCanbusPacketReceived(TPCANMsg msg)
    {
        if (CanbusPacketReceived != null)
        {
            CanbusPacketReceived(this, msg);
        }
    }

    public void WriteCanMsg(TPCANMsg msg)
    {
        try
        {
            PCANBasic.Write(_handle, ref msg);
        }
        catch (Exception ex)
        {
            Log.Error("Error writing CAN message: {Error}, Bytes: {Bytes}", ex.Message, String.Join(" ", msg.DATA));
            throw ex;
        }
    }

    private bool TryInitializePCANChannel(out TPCANHandle handle)
    {
        // SNIP
    }

    /// <summary>
    /// Close CAN connection
    /// </summary>
    /// <param name="handle">an open CAN connection</param>
    private void UninitializPCANChannel(TPCANHandle handle)
    {
        IsOpen = false;
        if(_canbusListenerThread.IsAlive && !_canbusListenerThread.Join(200))
        {
            _canbusListenerThread.Abort();
        }
        TPCANStatus status = PCANBasic.Uninitialize(handle);
        Log.Information("Uninitializing channel {Channel}: {Status}", handle, status);
    }

    public void Dispose()
    {
        if(IsOpen)
        {
            UninitializPCANChannel(_handle);
        }
    }

    public void Close()
    {
        this.Dispose();
    }
}

CommandBase.cs

All commands extend the abstractCommandBase. CommandBase has logic to monitor incoming CAN packets and parse them for a response for the given command. A given command has:

• Command specific properties
• Logic to encode the properties to a CAN message
• Logic to decode a byte array back into properties/basic types
• CommandBase defines several abstract methods that a command must implement so that CommandBase can parse the incoming CAN messages and identify the response for the given command (The first several bytes in a CAN message are used to identify the command/response)

When sending a command:

• The command encodes itself as a CAN message and passes it to the ExecuteCommandBase(..) method in CommandBase (This is called from the UI thread).
• CommandBase sends the packets, and subscribes to the CanbusConnection.CanbusPacketReceived event.
• The CanbusPacketReceived event handler is called from the background thread. I have a simple state machine that looks for for a matching set of first bytes for each incoming message. After that it collects packets and appends the payload to a list until the expected number of packets have been received.
• I use an AutoResetEvent with a timeout to synchronize between the background thread parsing and the UI thread.
• I use an extension method to convert the AutoResetEvent.WaitOne() to a Task<>

---

public abstract class CommandBase
{
    #region Parse Engine Data Structures
    private enum ParseState
    {
        NoCommand,
        CommandInProgress,
        CommandTimeout
    }

    private class ParseData
    {
        public List<byte> RxBytes { get; set; }
        public int NumPacketsReceived { get; set; }

        public int NumPacketsExpected { get; set; }

        public ParseData()
        {
            RxBytes = new List<byte>();
            NumPacketsReceived = 0;
            NumPacketsExpected = 1;
        }
    }
    #endregion

    private ParseState _parseState;
    private ParseData _parseData;
    private AutoResetEvent _parseWaitHandle;

    #region Properties

    /// <summary>
    /// Time in milliseconds before a timeout occurs
    /// </summary>
    public int ResponseTimeoutMilliseconds { get; protected set; }

    /// <summary>
    /// True is a timeout occurred
    /// </summary>
    public bool TimoutOccured { get; private set; }

    /// <summary>
    /// Gets the reply length
    /// </summary>
    /// <param name="firstPacketData">The first packet's data, for use with variable length responses</param>
    /// <returns>Number of packets required for the whole command</returns>
    protected abstract int GetReplyPacketLength(byte[] firstPacketData);
    #endregion


    #region Constructor
    protected CommandBase()
    {
        TimoutOccured = false;
        ResponseTimeoutMilliseconds = 500; // default timeout 500ms
    }
    #endregion


    protected async Task<bool> ExecuteCommandBase(CanbusConnection connection, TPCANMsg msg)
    {
        if(connection == null)
        {
            throw new ArgumentNullException("connection");
        }
        if(!connection.IsOpen)
        {
            throw new ArgumentException("Connection is not open");
        }

        _parseState = ParseState.NoCommand;
        _parseData = new ParseData();

        using (_parseWaitHandle = new AutoResetEvent(initialState: false))
        {

            lock (connection)
            {
                connection.CanbusPacketReceived += Connection_CanbusPacketReceived;
                connection.WriteCanMsg(msg);
            }

            bool allPacketsReceived = await _parseWaitHandle.WaitOneAsync(ResponseTimeoutMilliseconds);

            lock (connection)
            {
                connection.CanbusPacketReceived -= Connection_CanbusPacketReceived;
            }

            if (allPacketsReceived)
            {
                if(Parse(_parseData.RxBytes.ToArray()))
                {
                    Log.Information("Received {MsgType} message", this.GetType().ToString());
                    return true;
                }
                else
                {
                    Log.Information("Received malformed {MsgType} message", this.GetType().ToString());
                    return false;
                }
            }
            else
            {
                // timeout
                Log.Warning("{MsgType} message Timeout", this.GetType().ToString());
                TimoutOccured = true;
                return false;
            }
        }
    }

    /// <summary>
    /// Parses the incoming canbus packets, looking for a matching response start seqeunce.  Returns true if the start sequence is found
    /// </summary>
    /// <param name="firstPacketBytes">canbus packet data bytes</param>
    /// <returns></returns>
    protected abstract bool IsMatch(byte[] firstPacketBytes);

    /// <summary>
    /// Parses a byte sequence into the command response
    /// </summary>
    /// <param name="data">an array of bytes, guaranteed to be enough for the expcted response length</param>
    /// <returns>true if parse was successful, false if errors were encountered</returns>
    protected abstract bool Parse(byte[] data);

    // This function runs on (is called from) the CanbusConnection.CanusListenerThread, a background thread
    // The _parseWaitHandler is initialized on either the UI thread or a Task generated on the UI thread (I'm not quite clear on the underlying details)
    //   so this signals the waiting thread.  That thread is also configured with a timeout, so if parsing never completes, the calling thread
    //   should still be notified
    private void Connection_CanbusPacketReceived(object sender, TPCANMsg packet)
    {
        switch (_parseState)
        {
            case ParseState.NoCommand:
                if (IsMatch(packet.DATA))
                {
                    _parseData.NumPacketsExpected = GetReplyPacketLength(packet.DATA);
                    if (_parseData.NumPacketsExpected == 1)
                    {
                        _parseWaitHandle.Set();
                    }
                    else
                    {
                        _parseState = ParseState.CommandInProgress;
                        _parseData.RxBytes.AddRange(packet.DATA);
                        _parseData.NumPacketsReceived++;
                    }
                }
                break;

            case ParseState.CommandInProgress:
                _parseData.RxBytes.AddRange(packet.DATA);
                _parseData.NumPacketsReceived++;

                if (_parseData.NumPacketsReceived == _parseData.NumPacketsExpected)
                {
                    _parseWaitHandle.Set();
                }
                break;
        }
    }
}

Example Command - CmdGetRpm.cs

Here is an example command that gets the firmware version of one of the boards.

public class CmdMpuSerialNumber : CommandBase
{
    public string SerialNumber { get; protected set; }

    public async Task<bool> ExecuteCommand(CanbusConnection conn)
    {
        this.ResponseTimeoutMilliseconds = 100000;
        TPCANMsg msg = new TPCANMsg();
        msg.DATA = new byte[] { 0xAA, 0xAA, 0x08, 0x08, 0, 0, 0, 0 };
        msg.LEN = 8;
        msg.MSGTYPE = TPCANMessageType.PCAN_MESSAGE_EXTENDED;
        msg.ID = (uint)CanbusAddress.PC;

        return await this.ExecuteCommandBase(conn, msg);
    }

    #region CommandBase members
    static readonly byte[] RX_PATTERN = new byte[] { 0xAA, 0xAA, 0x08 };

    protected override int GetReplyPacketLength(byte[] firstPacketData)
    {
        return 5;
    }

    protected override bool IsMatch(byte[] firstPacketBytes)
    {
        int numItems = Math.Min(RX_PATTERN.Length, firstPacketBytes.Length);
        return firstPacketBytes.Take(numItems).SequenceEqual(RX_PATTERN.Take(numItems));
    }

    protected override bool Parse(byte[] data)
    {
        IEnumerable<byte> query = data.Skip(8).TakeWhile((currentByte) => currentByte != 0);
        byte[] substringBytes = query.ToArray<byte>();
        SerialNumber = System.Text.Encoding.Default.GetString(substringBytes);
        //SerialNumber = BitConverter.ToString(data, 8);  // the serial number starts at the begining of the second packet, byte 8
        return true;
    }
    #endregion
}

Example UI Code - Form1.cs

And here is an example of using the code in the GUI

public partial class Form1 : Form
{
    CanbusConnection _conn;
    //CommandParser _parser;

    public Form1()
    {
        InitializeComponent();
    }

    private void Form1_Load(object sender, EventArgs e)
    {
        try
        {
            _conn = new CanbusConnection();
            _conn.Open();
        }
        catch(Exception ex)
        {
            MessageBox.Show("ConnectionError: " + ex.ToString());
        }
    }


    private void Form1_FormClosed(object sender, FormClosedEventArgs e)
    {
        if(_conn != null)
        {
            _conn.Close();
            _conn = null;
        }
    }

    private async void button1_Click(object sender, EventArgs e)
    {
        CmdMpuSerialNumber cmd = new CmdMpuSerialNumber();
        bool result = await cmd.ExecuteCommand(_conn);
        MessageBox.Show(cmd.SerialNumber, "MPU SN");
    }
}

Final Thoughts

• I know the Thread.Sleep() has a bit of code smell, but I couldn't think of another way to do it. I briefly considered timers an a bunch of other stuff, but Thread.Sleep gets the same result without needing to instantiate a bunch of other objects.
• This was a big reference for me in designing this: Creating Async Libraries That Are Modular, Reusable and Fast, in Microsoft Visual C# and Visual Basic
• Comments on naming/syntax/style are appreciated, but comments on the design are appreciated more :)

Answer 1

Testability

This is my main concern when creating an API that talks to external references, in this case The CAN library. The problem with this library is that we are forced to consume a static class PCANbasic. There is no interface provided for us to work with. Testing our own code depends on a connection to the CAN bus.

// static class is a tight dependency, hard to mock out
PCANBasic.Write(_handle, ref msg);

Therefore, it is up to us to create an interface, an implementation using the static library and use a variable of this interface _canBusService in CanbusConnection. This way, we can

• mock the library for unit tests
• use a different version of the library (considering the signatures remain)

interface ICanService

public interface ICanService : IDisposable
{ 
    TPCANHandle Handle { get; }
    bool Initialize();
    TPCANStatus Read(out TPCANMsg message);
    void Write(ref TPCANMsg message);
}

implementation CanService

public class CanService : ICanService
{
    public TPCANHandle Handle { get; private set; }

    public bool Initialize()
    {
        // Handle = [redacted]
        return true;
    }

    public TPCANStatus Read(out TPCANMsg message)
    {
        return PCANBasic.Read(Handle, out message);
    }

    public void Read(ref TPCANMsg message)
    {
        PCANBasic.Write(Handle, ref message);
    }

    public void Dispose()
    {
        PCANBasic.Uninitialize(Handle);
        Handle = 0;
    }
}

---

Data Integrity

You have stated in the comments that the library is half-duplex, non-concurrent. This simplifies design, but still requires measures to enforce thread-safety.

Consider using a lock for Open and WriteCanMsg to avoid race conditions.

public object SyncRoot { get; } // .. create a new object() in the constructor

method Open

Since no lock is used, another thread can call the method after we think it is not opened.

if (IsOpen) { return true; }
// .. other code that assumes IsOpen == false

We no longer store the handle and use the newly created interface.

IsOpen = TryInitializePCANChannel(out _handle);

IsOpen = _canBusService.Initialize();

What happens if the call was not succesfull?

if(IsOpen)
{
    _canbusListenerThread = new Thread(new ThreadStart(CanbusListener));
    _canbusListenerThread.Start();
} // else ??

Invert the condition and trow on failure.

if (!IsOpen)
{
    throw new IOException("Failure initializing CAN bus");
}

Use throw to keep the original stacktrace.

catch(Exception ex)
{
    Log.Error("Error opening PCAN connection: {Message}, Error: {Error}", ex.Message, ex);
    throw ex;
}

catch(Exception ex)
{
    Log.Error("Error opening PCAN connection: {Message}, Error: {Error}", ex.Message, ex);
    throw;   // preserves stacktrace
}

refactor Open

public bool Open()
{
    lock (SyncRoot)
    {
        if (IsOpen) { return true; }

        try
        {
            IsOpen = _canBusService.Initialize();
            if (!IsOpen)
            {
                throw new IOException("Failure initializing CAN bus");
            }
            _canbusListenerThread = new Thread(new ThreadStart(CanbusListener));
            _canbusListenerThread.Start();
        }
        catch(Exception ex)
        {
            Log.Error("Error opening PCAN connection: {Message}, Error: {Error}", ex.Message, ex);
            throw;
        }
        return success;
    }
}

method WriteCanMsg

Guard against bad user input.

msg = msg ?? throw new ArgumentNullException(nameof(msg));

Same tips as with Open: use lock, interface, throw

refactor WriteCanMsg

public void WriteCanMsg(TPCANMsg msg)
{
    msg = msg ?? throw new ArgumentNullException(nameof(msg));
    lock (SyncRoot)
    {
        try
        {
            _canBusService.Write(ref msg);
        }
        catch (Exception ex)
        {
            Log.Error("Error writing CAN message: {Error}, Bytes: {Bytes}", ex.Message, String.Join(" ", msg.DATA));
            throw;
        }
    }
}

The commands should reuse the lock. For instance, CommandBase

 //lock (connection)
 lock (connection.SyncRoot)
 {
     connection.CanbusPacketReceived += Connection_CanbusPacketReceived;
     connection.WriteCanMsg(msg);
 }

refactor CanbusListener to use _canBusService

// ..
while(_canBusService.Read(out msg) == TPCANStatus.PCAN_ERROR_OK)
{
    // ..
}
// ..
Thread.Sleep(100); // <- I am surprised this is required. 
                   //    I would expect `_canBusService.Read` to be blocking.

---

Resource Management

Let Close do the actual work.

public void Close()
{  
    lock (SyncRoot) 
    {
        if (IsOpen)
        {
           _canBusService.Dispose();
           if(_canbusListenerThread.IsAlive && !_canbusListenerThread.Join(200))
           {
               _canbusListenerThread.Abort();
           }
        }
    }
}

Have Dispose calling Close. You could also refactor Dispose to use the dispose pattern. Clear event listeners and provide a destructor.

public void Dispose()
{
    Dispose(true);
    GC.SuppressFinalize(this);
}

~CanbusConnection()
{
    Dispose(false);
}

protected virtual void Dispose(bool disposing)
{
    if (disposing)
    {
        CanbusPacketReceived = null;
    }
    Close();
}

---

Separation of Concerns

• I would not expect CommandBase to have an operation protected async Task<bool> ExecuteCommandBase(CanbusConnection connection, TPCANMsg msg). This should be a method on CanbusConnection as public async Task<bool> ExecuteCommand(CommandBase command).
• Each specific command should provide a method BuildMessage() which would create the TPCANMsg from own state.

code CanbusConnection.ExecuteCommand

public class CanbusConnection
{
    // ..

    public async Task<bool> ExecuteCommand(CommandBase command)
    {
        var message = command.BuildMessage();
        // continue implementation with 'message' ..
    }
}

code CmdMpuSerialNumber.BuildMessage

public class CmdMpuSerialNumber : CommandBase
{
    // ..

    public TPCANMsg BuildMessage()
    {
        var msg = new TPCANMsg();
        msg.DATA = new byte[] { 0xAA, 0xAA, 0x08, 0x08, 0, 0, 0, 0 };
        msg.LEN = 8;
        msg.MSGTYPE = TPCANMessageType.PCAN_MESSAGE_EXTENDED;
        msg.ID = (uint)CanbusAddress.PC;
        return msg;
    }
}