非同期通信を基盤とするModbusサーバ設計

Modbus TCPサーバは複数クライアントの同時接続処理と正確なプロトコル解釈が求められます。主要設計要素は以下の通りです：

• 非同期I/Oによる接続処理
• スレッドセーフなデータ管理
• 拡張可能な機能コード実装
• イベント駆動型のデータ更新通知

C#による実装例

using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

public class ModbusTcpServer : IDisposable
{
    private TcpListener _serverSocket;
    private CancellationTokenSource _shutdownToken;
    private readonly object _storageLock = new object();
    
    // データストレージ
    private readonly Dictionary<byte, ushort[]> _holdingRegisters = new Dictionary<byte, ushort[]>();
    private readonly Dictionary<byte, bool[]> _coilStatuses = new Dictionary<byte, bool[]>();

    // イベント
    public event Action<byte, ushort, ushort> OnHoldingRegistersChanged;
    public event Action<byte, ushort, bool> OnCoilStatusChanged;

    public void StartServer(IPAddress bindAddress, int port = 502)
    {
        _shutdownToken = new CancellationTokenSource();
        _serverSocket = new TcpListener(bindAddress, port);
        _serverSocket.Start();
        Task.Run(() => AcceptConnections(_shutdownToken.Token));
    }

    public void StopServer()
    {
        _shutdownToken?.Cancel();
        _serverSocket?.Stop();
    }

    private async Task AcceptConnections(CancellationToken cancelToken)
    {
        while (!cancelToken.IsCancellationRequested)
        {
            try
            {
                var client = await _serverSocket.AcceptTcpClientAsync();
                _ = ProcessClientRequests(client, cancelToken);
            }
            catch (OperationCanceledException) { break; }
        }
    }

    private async Task ProcessClientRequests(TcpClient client, CancellationToken cancelToken)
    {
        using (client)
        {
            var networkStream = client.GetStream();
            var buffer = new byte[256];
            
            try
            {
                while (!cancelToken.IsCancellationRequested)
                {
                    int bytesRead = await networkStream.ReadAsync(buffer, 0, buffer.Length, cancelToken);
                    if (bytesRead == 0) break;

                    byte[] response = HandleModbusRequest(buffer.AsSpan(0, bytesRead).ToArray());
                    if (response != null)
                    {
                        await networkStream.WriteAsync(response, 0, response.Length, cancelToken);
                    }
                }
            }
            catch (IOException) { } // クライアント切断
        }
    }

    private byte[] HandleModbusRequest(byte[] requestFrame)
    {
        try
        {
            // MBAPヘッダー解析
            ushort transactionId = (ushort)((requestFrame[0] << 8) | requestFrame[1]);
            ushort protocolId = (ushort)((requestFrame[2] << 8) | requestFrame[3]);
            ushort frameLength = (ushort)((requestFrame[4] << 8) | requestFrame[5]);
            byte deviceId = requestFrame[6];
            
            if (protocolId != 0) return CreateErrorResponse(transactionId, deviceId, 0x04);

            byte[] pdu = new byte[frameLength - 1];
            Buffer.BlockCopy(requestFrame, 7, pdu, 0, pdu.Length);

            byte functionCode = pdu[0];
            
            return functionCode switch
            {
                0x01 => ProcessReadCoils(transactionId, deviceId, pdu),
                0x03 => ProcessReadRegisters(transactionId, deviceId, pdu),
                0x05 => ProcessWriteCoil(transactionId, deviceId, pdu),
                0x0F => ProcessWriteCoils(transactionId, deviceId, pdu),
                0x10 => ProcessWriteRegisters(transactionId, deviceId, pdu),
                _ => CreateErrorResponse(transactionId, deviceId, 0x01)
            };
        }
        catch
        {
            return CreateErrorResponse(0, 0, 0x04);
        }
    }

    private byte[] ProcessReadRegisters(ushort transactionId, byte deviceId, byte[] pdu)
    {
        ushort startAddr = (ushort)((pdu[1] << 8) | pdu[2]);
        ushort registerCount = (ushort)((pdu[3] << 8) | pdu[4]);

        lock (_storageLock)
        {
            if (!_holdingRegisters.TryGetValue(deviceId, out ushort[] registers))
                return CreateErrorResponse(transactionId, deviceId, 0x02);

            if (startAddr + registerCount > registers.Length)
                return CreateErrorResponse(transactionId, deviceId, 0x02);

            byte[] responsePdu = new byte[2 + registerCount * 2];
            responsePdu[0] = 0x03;
            responsePdu[1] = (byte)(registerCount * 2);

            for (int i = 0; i < registerCount; i++)
            {
                ushort value = registers[startAddr + i];
                responsePdu[2 + i * 2] = (byte)(value >> 8);
                responsePdu[3 + i * 2] = (byte)(value & 0xFF);
            }

            return CreateResponseFrame(transactionId, deviceId, responsePdu);
        }
    }

    private byte[] ProcessWriteRegisters(ushort transactionId, byte deviceId, byte[] pdu)
    {
        ushort startAddr = (ushort)((pdu[1] << 8) | pdu[2]);
        ushort registerCount = (ushort)((pdu[3] << 8) | pdu[4]);
        byte dataByteCount = pdu[5];

        lock (_storageLock)
        {
            if (!_holdingRegisters.ContainsKey(deviceId))
                return CreateErrorResponse(transactionId, deviceId, 0x02);

            if (startAddr + registerCount > _holdingRegisters[deviceId].Length)
                return CreateErrorResponse(transactionId, deviceId, 0x02);

            for (int i = 0; i < registerCount; i++)
            {
                int dataOffset = 6 + i * 2;
                ushort value = (ushort)((pdu[dataOffset] << 8) | pdu[dataOffset + 1]);
                _holdingRegisters[deviceId][startAddr + i] = value;
            }

            OnHoldingRegistersChanged?.Invoke(deviceId, startAddr, registerCount);

            byte[] responsePdu = new byte[5];
            responsePdu[0] = 0x10;
            responsePdu[1] = (byte)(startAddr >> 8);
            responsePdu[2] = (byte)(startAddr & 0xFF);
            responsePdu[3] = (byte)(registerCount >> 8);
            responsePdu[4] = (byte)(registerCount & 0xFF);
            return CreateResponseFrame(transactionId, deviceId, responsePdu);
        }
    }

    private byte[] CreateResponseFrame(ushort transactionId, byte deviceId, byte[] pdu)
    {
        byte[] frame = new byte[7 + pdu.Length];
        frame[0] = (byte)(transactionId >> 8);
        frame[1] = (byte)(transactionId & 0xFF);
        frame[2] = 0; // Protocol ID
        frame[3] = 0;
        frame[4] = (byte)((pdu.Length + 1) >> 8); // Length
        frame[5] = (byte)((pdu.Length + 1) & 0xFF);
        frame[6] = deviceId;
        Buffer.BlockCopy(pdu, 0, frame, 7, pdu.Length);
        return frame;
    }

    private byte[] CreateErrorResponse(ushort transactionId, byte deviceId, byte errorCode)
    {
        byte[] errorPdu = new byte[2];
        errorPdu[0] = (byte)(0x80 | errorCode);
        errorPdu[1] = errorCode;
        return CreateResponseFrame(transactionId, deviceId, errorPdu);
    }

    public void SetupDeviceRegisters(byte deviceId, ushort registerCount)
    {
        lock (_storageLock)
        {
            _holdingRegisters[deviceId] = new ushort[registerCount];
        }
    }

    public void SetupDeviceCoils(byte deviceId, ushort coilCount)
    {
        lock (_storageLock)
        {
            _coilStatuses[deviceId] = new bool[coilCount];
        }
    }

    public void Dispose()
    {
        StopServer();
        _serverSocket?.Stop();
        _shutdownToken?.Dispose();
    }
}

主要機能の特徴

• 非同期ネットワーク処理: AcceptConnectionsによるクライアント接続管理
• データ管理: デバイス単位のレジスタ/コイル初期化

  // デバイス1の保持レジスタを1000個で初期化
  server.SetupDeviceRegisters(deviceId: 1, registerCount: 1000);

• イベント通知: データ変更時のイベント発火

  server.OnHoldingRegistersChanged += (devId, startAddr, count) => 
  {
      Console.WriteLine($"Device {devId}: Registers {startAddr}-{startAddr+count-1} updated");
  };

• サポート機能コード: 0x01, 0x03, 0x05, 0x0F, 0x10
• エラー処理: 無効アドレス(0x02), 不正データ(0x03), サーバ障害(0x04)

拡張機能の実装例

// カスタムストレージプロバイダ
public interface IModbusDataRepository
{
    ushort[] GetHoldingRegisters(byte deviceId);
    bool[] GetCoilStatuses(byte deviceId);
    void UpdateHoldingRegisters(byte deviceId, ushort startAddress, ushort[] values);
    void UpdateCoilStatuses(byte deviceId, ushort startAddress, bool[] states);
}

// クライアント認証機能
private readonly HashSet<IPAddress> _authorizedClients = new HashSet<IPAddress>();

public bool AuthorizeClient(IPAddress clientAddress)
{
    return _authorizedClients.Contains(clientAddress);
}