文章/答案/技术大牛

发布

社区首页 >问答首页 >异步硬件通信-库设计

问异步硬件通信-库设计
EN

Code Review用户

提问于 2019-06-28 18:39:46

回答 1查看 212关注 0票数 3

我有一些通过CANbus进行通信的设备(“工具”)，我需要与它们进行通信。我使用的是提供基本.Net控制库的USB->CAN dongle .我正在编写一个抽象出工具交流的库，我想对设计提出一些意见。

请注意，我已经在.NET 2.0世界生活了很长时间了。我已经对.NET 4.5中的一些新线程概念做了一些速成课程风格的研究，但我确信我的代码是新旧风格的混合体。

以下是对该体系结构的简要概述：

CAN库(PCANbasic)允许扫描usb并打开连接。可以将can数据包写入连接，也可以轮询接收到的数据包。
该工具具有自定义通信方案，其中命令或响应可以由一个或多个can数据包组成。
我的库的目标是隐藏PCAN库。我希望能够打开一个连接(CanbusConnection)，并能够发送命令和接收命令响应(每个命令都是一个抽象类CommandBase的实例)。
所有的通信都必须在后台线程上进行，这样才不会拖住UI。

CanbusConnection.cs

这个类只是封装底层PCAN库来建立连接，然后启动一个后台线程，每100 to轮询传入的CAN消息。它公开一个CanbusPacketReceived事件。关闭CanbusConnection类将停止后台线程，然后关闭基础PCAN连接。

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
    }

    /// 
    /// Close CAN connection
    /// 
    /// an open CAN connection
    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

所有命令都扩展了抽象CommandBase。CommandBase具有监视传入CAN数据包并为给定命令的响应解析它们的逻辑。给定的命令有：

命令特定属性
将属性编码为CAN消息的逻辑。
将字节数组解码回属性/基本类型的逻辑
CommandBase定义了命令必须实现的几个抽象方法，以便CommandBase能够解析传入的can消息并识别给定命令的响应( CAN消息中的前几个字节用于标识命令/响应)。

发送命令时：

该命令将自身编码为CAN消息，并将其传递给CommandBase中的CommandBase方法(这是从UI线程调用的)。
CommandBase发送数据包，并订阅CanbusConnection.CanbusPacketReceived事件。
从后台线程调用CanbusPacketReceived事件处理程序。我有一个简单的状态机，它为每个传入的消息寻找一组匹配的第一个字节。在此之后，它收集数据包并将有效载荷附加到列表中，直到接收到预期的数据包数量。
我使用带有超时的AutoResetEvent在后台线程解析和UI线程之间进行同步。
我使用扩展方法将AutoResetEvent.WaitOne()转换为Task<>

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

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

        public int NumPacketsExpected { get; set; }

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

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

    #region Properties

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

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

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


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


    protected async Task 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;
            }
        }
    }

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

    /// 
    /// Parses a byte sequence into the command response
    /// 
    /// an array of bytes, guaranteed to be enough for the expcted response length
    /// true if parse was successful, false if errors were encountered
    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;
        }
    }
}

示例命令- CmdGetRpm.cs

下面是一个示例命令，它获取其中一个板的固件版本。

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

    public async Task 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 query = data.Skip(8).TakeWhile((currentByte) => currentByte != 0);
        byte[] substringBytes = query.ToArray();
        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
}

示例UI代码- Form1.cs

下面是一个在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");
    }
}

最后的想法

我知道Thread.Sleep()有一点代码味道，但我想不出另一种方法。我简要地考虑了定时器是一堆其他东西，但是Thread.Sleep得到了相同的结果，而不需要实例化一堆其他对象。
这是我在设计时的一个很大的参考：在MicrosoftVisualC#和Visual中创建模块化、可重用和快速的异步库
欢迎对命名/语法/样式的评论，但对设计的评论更受欢迎:)

multithreading

api

回答 1

Code Review用户

回答已采纳

发布于 2019-06-29 20:09:01

可测性

这是我在创建与外部引用(在本例中为CAN库)对话的API时所关心的主要问题。这个库的问题是我们被迫使用一个静态类PCANbasic。没有提供给我们使用的接口。测试我们自己的代码取决于到CAN总线的连接。

//静态类是一个紧密的依赖项，很难模拟出PCANBasic.Write(_handle，ref )；

因此，我们应该创建一个接口，一个使用静态库的实现，并在_canBusService中使用CanbusConnection中的一个变量。这样，我们可以

模拟库进行单元测试。
使用不同版本的库(考虑到签名仍然存在)

接口ICanService

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

实现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;
    }
}

数据完整性

您在评论中指出，图书馆是半双工的，非并发的.这简化了设计，但仍然需要加强线程安全的措施。

考虑为Open和WriteCanMsg使用锁以避免竞争条件。

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

方法Open

由于没有使用锁，所以当我们认为方法没有打开时，另一个线程可以调用它。

如果(IsOpen) {返回true；} // .其他假定IsOpen == false的代码

我们不再存储句柄并使用新创建的接口。

IsOpen = TryInitializePCANChannel(out _handle);

IsOpen = _canBusService.Initialize();

如果电话没有满，会发生什么？

if(IsOpen) { _canbusListenerThread =新线程(新ThreadStart(CanbusListener))；_canbusListenerThread.Start()；} // CanbusListener？

颠倒条件，处理失败。

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

使用throw保持原来的堆栈跟踪。

捕获(异常ex) {Log.Error(“错误打开PCAN连接：{Message}，Error：{Error}"，ex.Message，ex)；抛出ex；}

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

重构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;
    }
}

方法WriteCanMsg

防止用户输入错误。

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

与Open相同的提示:使用锁、接口、抛出

重构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;
        }
    }
}

这些命令应该重用锁。例如，CommandBase

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

重构CanbusListener以使用_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.

资源管理

让Close来做实际的工作。

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

让Dispose给Close打电话。还可以重构Dispose以使用dispose模式。清除事件侦听器并提供析构函数。

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

~CanbusConnection()
{
    Dispose(false);
}

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

关注点分离

我不希望CommandBase有一个操作protected async Task ExecuteCommandBase(CanbusConnection connection, TPCANMsg msg)。这应该是CanbusConnection作为public async Task ExecuteCommand(CommandBase command)的一个方法。
每个特定的命令都应该提供一个方法BuildMessage()，它将从自己的状态创建TPCANMsg。

代码CanbusConnection.ExecuteCommand

public class CanbusConnection
{
    // ..

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

代码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;
    }
}

票数 1

页面原文内容由Code Review提供。腾讯云小微IT领域专用引擎提供翻译支持

原文链接：

https://codereview.stackexchange.com/questions/223161

复制

相似问题

问异步硬件通信-库设计
EN

CanbusConnection.cs

CommandBase.cs

示例命令- CmdGetRpm.cs

示例UI代码- Form1.cs

最后的想法

回答 1

Code Review用户

可测性

数据完整性

资源管理

关注点分离

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问异步硬件通信-库设计EN

CanbusConnection.cs

CommandBase.cs

示例命令- CmdGetRpm.cs

示例UI代码- Form1.cs

最后的想法

回答 1

Code Review用户

可测性

数据完整性

资源管理

关注点分离

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问异步硬件通信-库设计
EN