I'm having an implementation using SerialPort in C# using Visual Studio 2019. I'm using it in a Windows Application, using .NET Framework 4.7.2.
When I open and initialize the SerialPort, I'm launching 2 "message pump tasks", one for receiving data and one for transmitting data.
The SerialPort has a ConcurrentQueue to store the commands to transmit, and empties that queue if it has commands available. If there are no more commands, the transmitting pump does nothing.
Each command sent to the external device will return an acknowledge (simply the string "A\n"). Other received data is considered as a command coming from the external device, and is consolidated in a shared ConcurrentQueue that stores all the received commands, ready to be processed by the main application.
My question is:
What is better:
- Using the asynchronous versions
BaseStream.ReadAsyncandBaseStream.WriteAsync - Using the synchronous versions
BaseStream.ReadandBaseStream.Write
Below are the 2 versions of my code. I don't see any difference in performance, and because the message pumps run in the background in both versions, I don't have the feeling that my UI thread is "blocking".
Synchronous version:
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO.Ports;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace SerialPortTest
{
class COMDevice
{
private readonly byte[] LF = { (byte)'\n' };
public List<string> _RegisteredCmds = new List<string>();
private SerialPort _serialPort = new SerialPort();
private string _DeviceName;
private string _ProcessorType;
private ConcurrentQueue<string> _RxQueue;
public ConcurrentQueue<string> _TxQueue = new ConcurrentQueue<string>();
private Task _RxPump;
private Task _TxPump;
private ManualResetEvent mreAck = new ManualResetEvent(false);
public bool IsOpen { get { return _serialPort.IsOpen; } }
public COMDevice(string portName, ConcurrentQueue<string> RxQueue, Int32 baudRate = 500000)
{
_serialPort.PortName = portName;
_serialPort.BaudRate = baudRate;
_RxQueue = RxQueue;
_serialPort.Handshake = Handshake.None;
// format 8-N-1
_serialPort.DataBits = 8;
_serialPort.Parity = Parity.None;
_serialPort.StopBits = StopBits.One;
_serialPort.NewLine = "\n";
_serialPort.ReadTimeout = 100;
_serialPort.WriteTimeout = 100;
}
public bool Open()
{
try
{
_serialPort.Open();
return true;
}
catch (TimeoutException)
{
if (_serialPort.IsOpen)
_serialPort.Close();
return false;
}
catch (Exception)
{
if (_serialPort.IsOpen)
_serialPort.Close();
return false;
}
}
public bool Close()
{
if (!_serialPort.IsOpen)
return false;
try
{
_serialPort.Write("RESET\n");
_serialPort.Close();
Task[] tasks = { _TxPump, _RxPump };
Task.WaitAll(tasks, 100);
return true;
}
catch (Exception)
{
return false;
}
}
private void ClearInputBuffer()
{
while (true)
{
int readCount = _serialPort.BytesToRead;
if (readCount == 0)
break;
byte[] buffer = new byte[readCount];
_serialPort.Read(buffer, 0, readCount);
}
}
public bool Initialize()
{
try
{
// clean receive buffer of Arduino and this application
_serialPort.Write("\n");
ClearInputBuffer();
// get identification
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: Write IDENT");
_serialPort.Write("IDENT\n");
_DeviceName = _serialPort.ReadLine();
_ProcessorType = _serialPort.ReadLine();
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: DeviceName: \"{_DeviceName}\" ProcessorType: \"{_ProcessorType}\"");
// register the LED/Data commands
_serialPort.Write("REGISTER\n");
while (true)
{
string sRegister = _serialPort.ReadLine();
if (sRegister == "")
break;
_RegisteredCmds.Add(sRegister);
}
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: {_RegisteredCmds.Count} Commands Registered");
// start transmitting and receiving commands
_TxPump = Task.Run(() => TxPump());
_RxPump = Task.Run(() => RxPump());
return true;
}
catch (Exception ex)
{
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: Exception {ex}");
return false;
}
}
private void RxPump()
{
var buffer = new byte[1024];
StringBuilder sbCmd = new StringBuilder("", 256);
try
{
while (_serialPort.IsOpen)
{
if (_serialPort.BytesToRead == 0)
continue;
int cnt = _serialPort.BaseStream.Read(buffer, 0, 1024);
for (int i = 0; i < cnt; i++)
{
if ((char)buffer[i] == '\n')
{
if ((sbCmd.Length == 1) && (sbCmd[0] == 'A'))
// ACK sequence received
mreAck.Set();
else if (sbCmd.Length != 0)
// Command received
_RxQueue.Enqueue(sbCmd.ToString());
sbCmd.Clear();
}
else
sbCmd.Append((char)buffer[i]);
}
}
}
catch (Exception ex)
{
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} RxCommandsAsync: Exception {ex}");
}
}
private void TxPump()
{
try
{
while (_serialPort.IsOpen)
{
string sCmd;
if (_TxQueue.Count == 0)
continue;
else
_TxQueue.TryDequeue(out sCmd);
byte[] buffer = Encoding.ASCII.GetBytes($"{sCmd}\n");
int attempts = 2;
while (attempts-- != 0)
{
_serialPort.BaseStream.Write(buffer, 0, sCmd.Length + 1);
// Reset the ManualResetEvent and wait for ACK for 50 msec
mreAck.Reset();
if (mreAck.WaitOne(50))
break;
else
{
Debug.WriteLine($"Not Ack {2 - attempts} for \"{sCmd}\"");
// Send linefeed to be sure we are in sync
_serialPort.BaseStream.Write(LF, 0, 1);
}
}
}
}
catch (Exception)
{
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} TxCommandsAsync: Exception");
}
}
}
}
Asynchronous version:
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO.Ports;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace SerialPortTest
{
class COMDevice
{
private readonly byte[] LF = { (byte)'\n' };
public List<string> _RegisteredCmds = new List<string>();
private SerialPort _serialPort = new SerialPort();
private string _DeviceName;
private string _ProcessorType;
private ConcurrentQueue<string> _RxQueue;
public ConcurrentQueue<string> _TxQueue = new ConcurrentQueue<string>();
private Task _RxPump;
private Task _TxPump;
private ManualResetEvent mreAck = new ManualResetEvent(false);
public bool IsOpen { get { return _serialPort.IsOpen; } }
public COMDevice(string portName, ConcurrentQueue<string> RxQueue, Int32 baudRate = 500000)
{
_serialPort.PortName = portName;
_serialPort.BaudRate = baudRate;
_RxQueue = RxQueue;
_serialPort.Handshake = Handshake.None;
// format 8-N-1
_serialPort.DataBits = 8;
_serialPort.Parity = Parity.None;
_serialPort.StopBits = StopBits.One;
_serialPort.NewLine = "\n";
_serialPort.ReadTimeout = 100;
_serialPort.WriteTimeout = 100;
}
public bool Open()
{
try
{
_serialPort.Open();
return true;
}
catch (TimeoutException)
{
if (_serialPort.IsOpen)
_serialPort.Close();
return false;
}
catch (Exception)
{
if (_serialPort.IsOpen)
_serialPort.Close();
return false;
}
}
public bool Close()
{
if (!_serialPort.IsOpen)
return false;
try
{
_serialPort.Write("RESET\n");
_serialPort.Close();
Task[] tasks = { _TxPump, _RxPump };
Task.WaitAll(tasks, 100);
return true;
}
catch (Exception)
{
return false;
}
}
private void ClearInputBuffer()
{
while (true)
{
int readCount = _serialPort.BytesToRead;
if (readCount == 0)
break;
byte[] buffer = new byte[readCount];
_serialPort.Read(buffer, 0, readCount);
}
}
public bool Initialize()
{
try
{
// clean receive buffer of Arduino and this application
_serialPort.Write("\n");
ClearInputBuffer();
// get identification
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: Write IDENT");
_serialPort.Write("IDENT\n");
_DeviceName = _serialPort.ReadLine();
_ProcessorType = _serialPort.ReadLine();
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: DeviceName: \"{_DeviceName}\" ProcessorType: \"{_ProcessorType}\"");
// register the LED/Data commands
_serialPort.Write("REGISTER\n");
while (true)
{
string sRegister = _serialPort.ReadLine();
if (sRegister == "")
break;
_RegisteredCmds.Add(sRegister);
}
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: {_RegisteredCmds.Count} Commands Registered");
// start transmitting and receiving commands
_TxPump = Task.Run(() => TxPumpAsync());
_RxPump = Task.Run(() => RxPumpAsync());
return true;
}
catch (Exception ex)
{
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} Initialize: Exception {ex}");
return false;
}
}
private async Task RxPumpAsync()
{
var buffer = new byte[1024];
StringBuilder sbCmd = new StringBuilder("", 256);
try
{
while (_serialPort.IsOpen)
{
int cnt = await _serialPort.BaseStream.ReadAsync(buffer, 0, 1024);
for (int i = 0; i < cnt; i++)
{
if ((char)buffer[i] == '\n')
{
if ((sbCmd.Length == 1) && (sbCmd[0] == 'A'))
// ACK sequence received
mreAck.Set();
else if (sbCmd.Length != 0)
// Command received
_RxQueue.Enqueue(sbCmd.ToString());
sbCmd.Clear();
}
else
sbCmd.Append((char)buffer[i]);
}
}
}
catch (Exception ex)
{
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} RxCommandsAsync: Exception {ex}");
}
}
private async Task TxPumpAsync()
{
try
{
while (_serialPort.IsOpen)
{
string sCmd;
if (_TxQueue.Count == 0)
continue;
else
_TxQueue.TryDequeue(out sCmd);
byte[] buffer = Encoding.ASCII.GetBytes($"{sCmd}\n");
int attempts = 2;
while (attempts-- != 0)
{
await _serialPort.BaseStream.WriteAsync(buffer, 0, sCmd.Length + 1);
// Reset the ManualResetEvent and wait for ACK for 50 msec
mreAck.Reset();
if (mreAck.WaitOne(50))
break;
else
{
Debug.WriteLine($"Not Ack {2 - attempts} for \"{sCmd}\"");
// Send linefeed to be sure we are in sync
await _serialPort.BaseStream.WriteAsync(LF, 0, 1);
}
}
}
}
catch (Exception)
{
Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss:fff")} TxCommandsAsync: Exception");
}
}
}
}
