Initial commit
This commit is contained in:
@@ -0,0 +1,556 @@
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using RobotNet10.CANOpen.Interfaces;
|
||||
using RobotNet10.CANOpen.Services;
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace RobotNet10.CANOpen;
|
||||
|
||||
/// <summary>
|
||||
/// Implementation của ICanOpenManager và IHostedService
|
||||
/// Quản lý ICanBus và CanOpenDevice instances với thread-safe operations
|
||||
/// Tự động tạo SocketCanBus từ interface name
|
||||
/// </summary>
|
||||
public class CanOpenManager : ICanOpenManager, IHostedService
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, ICanBus> _canBuses = [];
|
||||
private readonly ConcurrentDictionary<string, Dictionary<byte, CanOpenDevice>> _devices = []; // Key: interfaceName -> (nodeId -> device)
|
||||
private readonly SemaphoreSlim _semaphore = new(1, 1); // Binary semaphore for exclusive access
|
||||
private readonly object _stoppingLock = new(); // Lock object for _stopping flag
|
||||
private readonly ILoggerFactory? _loggerFactory;
|
||||
private readonly ILogger<CanOpenManager>? _logger;
|
||||
private readonly TimeSpan _lockTimeout = TimeSpan.FromSeconds(30); // Timeout cho lock operations
|
||||
private readonly int _maxRetryCount; // Số lần retry khi acquire lock thất bại
|
||||
private readonly TimeSpan _retryDelay; // Delay giữa các lần retry
|
||||
private bool _disposed;
|
||||
private bool _stopping;
|
||||
|
||||
/// <summary>
|
||||
/// Constructor với các tham số cấu hình
|
||||
/// </summary>
|
||||
/// <param name="loggerFactory">Logger factory để tạo logger</param>
|
||||
/// <param name="maxRetryCount">Số lần retry khi acquire lock thất bại (mặc định: 3)</param>
|
||||
/// <param name="lockTimeout">Timeout cho mỗi lần acquire lock (mặc định: 30 giây)</param>
|
||||
/// <param name="retryDelay">Delay giữa các lần retry (mặc định: 100ms)</param>
|
||||
public CanOpenManager(
|
||||
ILoggerFactory? loggerFactory = null,
|
||||
int maxRetryCount = 3,
|
||||
TimeSpan? lockTimeout = null,
|
||||
TimeSpan? retryDelay = null)
|
||||
{
|
||||
_loggerFactory = loggerFactory;
|
||||
_logger = loggerFactory?.CreateLogger<CanOpenManager>();
|
||||
_maxRetryCount = maxRetryCount >= 0 ? maxRetryCount : throw new ArgumentException("MaxRetryCount must be >= 0", nameof(maxRetryCount));
|
||||
_lockTimeout = lockTimeout ?? TimeSpan.FromSeconds(30);
|
||||
_retryDelay = retryDelay ?? TimeSpan.FromMilliseconds(100);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Acquire lock với timeout và retry để tránh deadlock
|
||||
/// </summary>
|
||||
private async Task<IDisposable> AcquireLockAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
Exception? lastException = null;
|
||||
|
||||
for (int attempt = 0; attempt <= _maxRetryCount; attempt++)
|
||||
{
|
||||
// Nếu không phải lần thử đầu tiên, đợi một chút trước khi retry
|
||||
if (attempt > 0)
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(_retryDelay, cancellationToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
|
||||
{
|
||||
cts.CancelAfter(_lockTimeout);
|
||||
|
||||
try
|
||||
{
|
||||
await _semaphore.WaitAsync(cts.Token);
|
||||
return new SemaphoreRelease(_semaphore);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (OperationCanceledException ex)
|
||||
{
|
||||
lastException = new TimeoutException(
|
||||
$"Failed to acquire lock within {_lockTimeout.TotalSeconds} seconds (attempt {attempt + 1}/{_maxRetryCount + 1})",
|
||||
ex);
|
||||
|
||||
// Nếu đã hết số lần retry, throw exception
|
||||
if (attempt >= _maxRetryCount)
|
||||
{
|
||||
throw lastException;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Không bao giờ đến đây, nhưng compiler cần return statement
|
||||
throw lastException ?? new TimeoutException($"Failed to acquire lock after {_maxRetryCount + 1} attempts");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Acquire lock với timeout và retry (synchronous version)
|
||||
/// </summary>
|
||||
private IDisposable AcquireLock()
|
||||
{
|
||||
Exception? lastException = null;
|
||||
|
||||
for (int attempt = 0; attempt <= _maxRetryCount; attempt++)
|
||||
{
|
||||
// Nếu không phải lần thử đầu tiên, đợi một chút trước khi retry
|
||||
if (attempt > 0)
|
||||
{
|
||||
Thread.Sleep(_retryDelay);
|
||||
}
|
||||
|
||||
if (_semaphore.Wait(_lockTimeout))
|
||||
{
|
||||
return new SemaphoreRelease(_semaphore);
|
||||
}
|
||||
|
||||
lastException = new TimeoutException(
|
||||
$"Failed to acquire lock within {_lockTimeout.TotalSeconds} seconds (attempt {attempt + 1}/{_maxRetryCount + 1})");
|
||||
|
||||
// Nếu đã hết số lần retry, throw exception
|
||||
if (attempt >= _maxRetryCount)
|
||||
{
|
||||
throw lastException;
|
||||
}
|
||||
}
|
||||
|
||||
// Không bao giờ đến đây, nhưng compiler cần return statement
|
||||
throw lastException ?? new TimeoutException($"Failed to acquire lock after {_maxRetryCount + 1} attempts");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Helper class để release semaphore khi dispose
|
||||
/// </summary>
|
||||
private sealed class SemaphoreRelease : IDisposable
|
||||
{
|
||||
private readonly SemaphoreSlim _semaphore;
|
||||
private bool _disposed;
|
||||
|
||||
public SemaphoreRelease(SemaphoreSlim semaphore)
|
||||
{
|
||||
_semaphore = semaphore;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (!_disposed)
|
||||
{
|
||||
_semaphore.Release();
|
||||
_disposed = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<ICanBus> GetOrCreateCanBusAsync(string interfaceName, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
throw new ObjectDisposedException(nameof(CanOpenManager), "CanOpenManager is disposed or stopping");
|
||||
|
||||
if (string.IsNullOrWhiteSpace(interfaceName))
|
||||
throw new ArgumentException("Interface name cannot be null or empty", nameof(interfaceName));
|
||||
|
||||
// Chưa có trong dictionary, cần tạo mới
|
||||
// Dùng lock để đảm bảo chỉ một thread tạo và connect
|
||||
using (await AcquireLockAsync(cancellationToken))
|
||||
{
|
||||
// Double-check: có thể đã được tạo bởi thread khác trong lúc chờ lock
|
||||
if (_canBuses.TryGetValue(interfaceName, out var existingBus))
|
||||
{
|
||||
return existingBus;
|
||||
}
|
||||
|
||||
// Tạo mới SocketCanBus instance
|
||||
var canBusLogger = _loggerFactory?.CreateLogger<SocketCanBus>();
|
||||
var canBus = new SocketCanBus(interfaceName, canBusLogger);
|
||||
|
||||
// TODO: Shutdown và enable lại interfaceName
|
||||
|
||||
try
|
||||
{
|
||||
// Connect đến CAN bus
|
||||
await canBus.ConnectAsync(cancellationToken);
|
||||
|
||||
// Add vào dictionary
|
||||
if (_canBuses.TryAdd(interfaceName, canBus))
|
||||
{
|
||||
return canBus;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Should not happen vì đã check và lock, nhưng handle để an toàn
|
||||
// Nếu vẫn add thất bại, có thể thread khác đã add trong lúc này
|
||||
if (_canBuses.TryGetValue(interfaceName, out existingBus))
|
||||
{
|
||||
_logger?.LogWarning("Another thread added CAN bus for interface {InterfaceName} during creation", interfaceName);
|
||||
// Dispose instance vừa tạo vì không thể add vào dictionary
|
||||
try
|
||||
{
|
||||
await canBus.DisconnectAsync(cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger?.LogWarning(ex, "Error disconnecting CAN bus during cleanup");
|
||||
}
|
||||
canBus.Dispose();
|
||||
return existingBus;
|
||||
}
|
||||
|
||||
_logger?.LogError("Failed to add CAN bus to dictionary for interface {InterfaceName}", interfaceName);
|
||||
throw new InvalidOperationException($"Failed to add CAN bus for interface {interfaceName}");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Nếu connect fail hoặc có lỗi khác, dispose instance đã tạo
|
||||
_logger?.LogError(ex, "Error creating CAN bus for interface {InterfaceName}", interfaceName);
|
||||
try
|
||||
{
|
||||
if (canBus.IsConnected)
|
||||
{
|
||||
await canBus.DisconnectAsync(cancellationToken);
|
||||
}
|
||||
}
|
||||
catch (Exception disconnectEx)
|
||||
{
|
||||
_logger?.LogWarning(disconnectEx, "Error disconnecting CAN bus during error cleanup");
|
||||
}
|
||||
canBus.Dispose();
|
||||
throw;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public ICanBus? GetCanBus(string interfaceName)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
throw new ObjectDisposedException(nameof(CanOpenManager), "CanOpenManager is disposed or stopping");
|
||||
|
||||
// ConcurrentDictionary.TryGetValue is thread-safe, no lock needed
|
||||
return _canBuses.TryGetValue(interfaceName, out var canBus) ? canBus : null;
|
||||
}
|
||||
|
||||
public async Task<CanOpenDevice> GetOrCreateDeviceAsync(string interfaceName, byte nodeId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
throw new ObjectDisposedException(nameof(CanOpenManager), "CanOpenManager is disposed or stopping");
|
||||
|
||||
if (nodeId == 0 || nodeId > 127)
|
||||
throw new ArgumentException("NodeId must be between 1 and 127", nameof(nodeId));
|
||||
|
||||
// Đảm bảo ICanBus đã được tạo và connect
|
||||
var canBus = await GetOrCreateCanBusAsync(interfaceName, cancellationToken);
|
||||
|
||||
// Với ConcurrentDictionary, cần lock cho inner Dictionary<byte, CanOpenDevice>
|
||||
// Vì inner Dictionary không phải thread-safe
|
||||
using (await AcquireLockAsync(cancellationToken))
|
||||
{
|
||||
// Kiểm tra xem device đã tồn tại chưa
|
||||
if (_devices.TryGetValue(interfaceName, out var devicesForInterface))
|
||||
{
|
||||
if (devicesForInterface.TryGetValue(nodeId, out var existingDevice))
|
||||
{
|
||||
return existingDevice;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Tạo mới inner dictionary nếu chưa có
|
||||
devicesForInterface = [];
|
||||
_devices[interfaceName] = devicesForInterface;
|
||||
}
|
||||
|
||||
// Tạo mới CanOpenDevice với logger
|
||||
var deviceLogger = _loggerFactory?.CreateLogger<CanOpenDevice>();
|
||||
var device = new CanOpenDevice(canBus, nodeId, deviceLogger, _loggerFactory);
|
||||
devicesForInterface[nodeId] = device;
|
||||
return device;
|
||||
}
|
||||
}
|
||||
|
||||
public CanOpenDevice? GetDevice(string interfaceName, byte nodeId)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
throw new ObjectDisposedException(nameof(CanOpenManager), "CanOpenManager is disposed or stopping");
|
||||
|
||||
using (AcquireLock())
|
||||
{
|
||||
if (_devices.TryGetValue(interfaceName, out var devicesForInterface))
|
||||
{
|
||||
return devicesForInterface.TryGetValue(nodeId, out var device) ? device : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public bool RemoveDevice(string interfaceName, byte nodeId)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
return false;
|
||||
|
||||
using (AcquireLock())
|
||||
{
|
||||
if (!_devices.TryGetValue(interfaceName, out var devicesForInterface))
|
||||
return false;
|
||||
|
||||
if (!devicesForInterface.TryGetValue(nodeId, out var device))
|
||||
return false;
|
||||
|
||||
// Dispose device
|
||||
device.Dispose();
|
||||
devicesForInterface.Remove(nodeId);
|
||||
|
||||
// Nếu không còn device nào cho interface này, xóa dictionary
|
||||
if (devicesForInterface.Count == 0)
|
||||
{
|
||||
_devices.Remove(interfaceName, out var _);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<bool> RemoveCanBusAsync(string interfaceName, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
return false;
|
||||
|
||||
List<CanOpenDevice> devicesToDispose = [];
|
||||
ICanBus? canBus = null;
|
||||
|
||||
using (await AcquireLockAsync(cancellationToken))
|
||||
{
|
||||
// Check và collect trong cùng một lock để tránh race condition
|
||||
if (!_canBuses.TryGetValue(interfaceName, out canBus))
|
||||
return false;
|
||||
|
||||
// Collect tất cả devices đang sử dụng canBus này
|
||||
if (_devices.TryGetValue(interfaceName, out var devicesForInterface))
|
||||
{
|
||||
devicesToDispose.AddRange(devicesForInterface.Values);
|
||||
_devices.Remove(interfaceName, out var _);
|
||||
}
|
||||
|
||||
_canBuses.Remove(interfaceName, out var _);
|
||||
}
|
||||
|
||||
// Dispose devices trước (ngoài lock)
|
||||
foreach (var device in devicesToDispose)
|
||||
{
|
||||
device.Dispose();
|
||||
}
|
||||
|
||||
// Disconnect và dispose canBus (ngoài lock)
|
||||
await canBus.DisconnectAsync(cancellationToken);
|
||||
canBus.Dispose();
|
||||
return true;
|
||||
}
|
||||
|
||||
public IReadOnlyList<string> GetManagedInterfaces()
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
return Array.Empty<string>().AsReadOnly();
|
||||
|
||||
// ConcurrentDictionary.Keys is thread-safe for enumeration
|
||||
return _canBuses.Keys.ToList().AsReadOnly();
|
||||
}
|
||||
|
||||
public IReadOnlyList<byte> GetManagedNodeIds(string interfaceName)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
return Array.Empty<byte>().AsReadOnly();
|
||||
|
||||
// Need lock because inner Dictionary<byte, CanOpenDevice> is not thread-safe
|
||||
using (AcquireLock())
|
||||
{
|
||||
if (!_devices.TryGetValue(interfaceName, out var devicesForInterface))
|
||||
return Array.Empty<byte>().AsReadOnly();
|
||||
|
||||
return devicesForInterface.Keys.ToList().AsReadOnly();
|
||||
}
|
||||
}
|
||||
|
||||
public bool IsInterfaceManaged(string interfaceName)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
return false;
|
||||
|
||||
// ConcurrentDictionary.ContainsKey is thread-safe
|
||||
return _canBuses.ContainsKey(interfaceName);
|
||||
}
|
||||
|
||||
public bool IsDeviceManaged(string interfaceName, byte nodeId)
|
||||
{
|
||||
if (IsStoppingOrDisposed())
|
||||
return false;
|
||||
|
||||
// Need lock because inner Dictionary<byte, CanOpenDevice> is not thread-safe
|
||||
using (AcquireLock())
|
||||
{
|
||||
if (!_devices.TryGetValue(interfaceName, out var devicesForInterface))
|
||||
return false;
|
||||
|
||||
return devicesForInterface.ContainsKey(nodeId);
|
||||
}
|
||||
}
|
||||
|
||||
#region IHostedService
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
// Mark as stopping to prevent new operations
|
||||
lock (_stoppingLock)
|
||||
{
|
||||
_stopping = true;
|
||||
}
|
||||
|
||||
await CleanupResourcesAsync(cancellationToken);
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region IDisposable
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed)
|
||||
return;
|
||||
|
||||
// Mark as stopping
|
||||
lock (_stoppingLock)
|
||||
{
|
||||
_stopping = true;
|
||||
}
|
||||
|
||||
// Cleanup resources synchronously (for Dispose, we use blocking approach)
|
||||
try
|
||||
{
|
||||
var cleanupTask = CleanupResourcesAsync(CancellationToken.None);
|
||||
if (!cleanupTask.Wait(TimeSpan.FromSeconds(30)))
|
||||
{
|
||||
_logger?.LogWarning("Timeout waiting for cleanup to complete during Dispose");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger?.LogError(ex, "Error during cleanup in Dispose");
|
||||
}
|
||||
|
||||
_semaphore.Dispose();
|
||||
_disposed = true;
|
||||
GC.SuppressFinalize(this);
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region Private Helper Methods
|
||||
|
||||
/// <summary>
|
||||
/// Kiểm tra xem manager có đang stopping hoặc disposed không (thread-safe)
|
||||
/// </summary>
|
||||
private bool IsStoppingOrDisposed()
|
||||
{
|
||||
lock (_stoppingLock)
|
||||
{
|
||||
return _disposed || _stopping;
|
||||
}
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region Private Cleanup Methods
|
||||
|
||||
/// <summary>
|
||||
/// Cleanup tất cả resources (devices và canBuses) một cách async
|
||||
/// </summary>
|
||||
private async Task CleanupResourcesAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
List<CanOpenDevice> devicesToDispose = [];
|
||||
List<ICanBus> canBusesToDispose = [];
|
||||
|
||||
// Collect tất cả resources cần dispose (trong lock ngắn)
|
||||
using (await AcquireLockAsync(cancellationToken))
|
||||
{
|
||||
// Collect devices
|
||||
foreach (var devicesForInterface in _devices.Values)
|
||||
{
|
||||
devicesToDispose.AddRange(devicesForInterface.Values);
|
||||
}
|
||||
_devices.Clear();
|
||||
|
||||
// Collect canBuses
|
||||
canBusesToDispose.AddRange(_canBuses.Values);
|
||||
_canBuses.Clear();
|
||||
}
|
||||
|
||||
// Dispose devices (ngoài lock để tránh blocking)
|
||||
foreach (var device in devicesToDispose)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
try
|
||||
{
|
||||
device.Dispose();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger?.LogWarning(ex, "Error disposing CanOpenDevice");
|
||||
}
|
||||
}
|
||||
|
||||
// Disconnect và dispose canBuses (ngoài lock để tránh blocking async calls)
|
||||
foreach (var canBus in canBusesToDispose)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
try
|
||||
{
|
||||
// Disconnect với timeout
|
||||
using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
|
||||
{
|
||||
cts.CancelAfter(TimeSpan.FromSeconds(5));
|
||||
await canBus.DisconnectAsync(cts.Token);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger?.LogWarning("Timeout or cancellation while disconnecting CAN bus");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger?.LogWarning(ex, "Error disconnecting CAN bus");
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
canBus.Dispose();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger?.LogWarning(ex, "Error disposing CAN bus");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#endregion
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user