工业物联网采集开发：Modbus TCP传感器多并发连接的服务端实现

MBAP 头（7 字节）：
┌───────────────┬───────────────┬───────────────┬───────────────┐
│ 事务标识符    │ 协议标识符    │ 长度          │ 单元标识符    │
│ (2 bytes)     │ (2 bytes)     │ (2 bytes)     │ (1 byte)      │
│ Transaction   │ Protocol      │ Length        │ Unit ID       │
│ ID            │ ID (=0)       │               │               │
├───────────────┴───────────────┴───────────────┴───────────────┤
│ PDU（协议数据单元）：功能码(1 byte) + 数据(N bytes)           │
└─────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│                  主线程 / 事件循环                        │
│  ┌─────────┐  ┌──────────┐  ┌──────────────────────┐   │
│  │ Listener │  │ Acceptor │  │ 连接管理器           │   │
│  │ Socket   │─▶│ (accept) │─▶│ (fd表/超时/限流)     │   │
│  └─────────┘  └──────────┘  └──────────┬───────────┘   │
│                                         │               │
│  ┌──────────────────────────────────────▼───────────┐   │
│  │            I/O 多路复用层 (epoll/select)          │   │
│  │  监听所有已连接 socket 的读/写/异常事件            │   │
│  └──────────────────────────┬────────────────────────┘   │
│                             │                           │
│  ┌──────────────────────────▼────────────────────────┐   │
│  │           协议解析与分发层                         │   │
│  │  MBAP解析 → 功能码分发 → 寄存器操作 → 构造响应    │   │
│  └──────────────────────────┬────────────────────────┘   │
│                             │                           │
│  ┌──────────────────────────▼────────────────────────┐   │
│  │           寄存器映射层（共享内存/数据结构）          │   │
│  │  线圈表 / 离散输入表 / 输入寄存器 / 保持寄存器     │   │
│  │  读写锁保护，支持多线程并发访问                     │   │
│  └───────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────┘

# modbus_server_core.py
import struct
import socket
import select
import threading
from collections import defaultdict
from enum import IntEnum

class FunctionCode(IntEnum):
    READ_COILS = 0x01
    READ_DISCRETE_INPUTS = 0x02
    READ_HOLDING_REGISTERS = 0x03
    READ_INPUT_REGISTERS = 0x04
    WRITE_SINGLE_COIL = 0x05
    WRITE_SINGLE_REGISTER = 0x06
    WRITE_MULTIPLE_COILS = 0x0F
    WRITE_MULTIPLE_REGISTERS = 0x10

class ExceptionCode(IntEnum):
    ILLEGAL_FUNCTION = 0x01
    ILLEGAL_DATA_ADDRESS = 0x02
    ILLEGAL_DATA_VALUE = 0x03
    SLAVE_DEVICE_FAILURE = 0x04
    ACKNOWLEDGE = 0x05
    SLAVE_DEVICE_BUSY = 0x06
    MEMORY_PARITY_ERROR = 0x08
    GATEWAY_PATH_UNAVAILABLE = 0x0A
    GATEWAY_TARGET_NO_RESPONSE = 0x0B

class RegisterMap:
    """寄存器映射表（线程安全）"""
    def __init__(self, unit_id=1):
        self.unit_id = unit_id
        self._coils = bytearray(1024)          # 1024 线圈 (DO)
        self._discrete_inputs = bytearray(1024) # 1024 离散输入 (DI)
        self._input_registers = [0] * 1024      # 1024 输入寄存器 (只读)
        self._holding_registers = [0] * 1024    # 1024 保持寄存器 (可读写)
        self._lock = threading.RLock()

    def read_coils(self, start_addr, count):
        with self._lock:
            if start_addr + count > len(self._coils):
                raise ValueError("Address out of range")
            result = []
            for i in range(start_addr, start_addr + count):
                byte_idx = i // 8
                bit_idx = i % 8
                result.append(bool((self._coils[byte_idx] >> bit_idx) & 1))
            return result

    def write_coil(self, addr, value):
        with self._lock:
            if addr >= len(self._coils) * 8:
                raise ValueError("Address out of range")
            byte_idx = addr // 8
            bit_idx = addr % 8
            if value:
                self._coils[byte_idx] |= (1 << bit_idx)
            else:
                self._coils[byte_idx] &= ~(1 << bit_idx)

    def read_holding_registers(self, start_addr, count):
        with self._lock:
            if start_addr + count > len(self._holding_registers):
                raise ValueError("Address out of range")
            return self._holding_registers[start_addr:start_addr + count]

    def write_holding_register(self, addr, value):
        with self._lock:
            if addr >= len(self._holding_registers):
                raise ValueError("Address out of range")
            self._holding_registers[addr] = value & 0xFFFF

    def write_holding_registers(self, start_addr, values):
        with self._lock:
            if start_addr + len(values) > len(self._holding_registers):
                raise ValueError("Address out of range")
            for i, v in enumerate(values):
                self._holding_registers[start_addr + i] = v & 0xFFFF

    def read_input_registers(self, start_addr, count):
        with self._lock:
            if start_addr + count > len(self._input_registers):
                raise ValueError("Address out of range")
            return self._input_registers[start_addr:start_addr + count]

    def update_input_register(self, addr, value):
        """更新输入寄存器（模拟传感器数据变化）"""
        with self._lock:
            if addr >= len(self._input_registers):
                raise ValueError("Address out of range")
            self._input_registers[addr] = value & 0xFFFF

# modbus_tcp_server_select.py
import select
import struct
import socket
import threading
import time
import logging
from modbus_server_core import FunctionCode, ExceptionCode, RegisterMap

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)

class ModbusTCPServer:
    def __init__(self, host='0.0.0.0', port=502, max_connections=50):
        self.host = host
        self.port = port
        self.max_connections = max_connections
        self.register_maps = {}          # unit_id -> RegisterMap
        self._running = False
        self._server_sock = None
        self._connections = {}           # fd -> (socket, addr, last_active)
        self._recv_buffers = {}          # fd -> bytearray (处理粘包)
        self._lock = threading.Lock()

    def add_register_map(self, unit_id, reg_map):
        self.register_maps[unit_id] = reg_map

    def start(self):
        self._server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self._server_sock.bind((self.host, self.port))
        self._server_sock.listen(self.max_connections)
        self._server_sock.setblocking(False)
        self._running = True

        logger.info(f"Modbus TCP Server listening on {self.host}:{self.port}")
        logger.info(f"Max connections: {self.max_connections}")

        self._event_loop()

    def stop(self):
        self._running = False
        if self._server_sock:
            self._server_sock.close()
        with self._lock:
            for fd, (sock, addr, _) in self._connections.items():
                try:
                    sock.close()
                except:
                    pass
            self._connections.clear()

    def _event_loop(self):
        """主事件循环，使用 select 多路复用"""
        while self._running:
            with self._lock:
                read_fds = [self._server_sock] + [sock for sock, _, _ in self._connections.values()]
                write_fds = []
                error_fds = read_fds[:]

            try:
                readable, writable, exceptional = select.select(read_fds, write_fds, error_fds, 1.0)
            except select.error as e:
                logger.error(f"select error: {e}")
                continue

            # 处理新连接
            if self._server_sock in readable:
                self._accept_new_connection()
                readable.remove(self._server_sock)

            # 处理已连接 socket 的数据
            for sock in readable:
                fd = sock.fileno()
                try:
                    self._handle_client_data(fd, sock)
                except Exception as e:
                    logger.error(f"Error handling client {fd}: {e}")
                    self._close_connection(fd)

            # 处理异常
            for sock in exceptional:
                fd = sock.fileno()
                logger.warning(f"Exception on fd {fd}")
                self._close_connection(fd)

            # 定期清理超时连接
            self._cleanup_stale_connections()

    def _accept_new_connection(self):
        try:
            client_sock, client_addr = self._server_sock.accept()
            fd = client_sock.fileno()

            if len(self._connections) >= self.max_connections:
                logger.warning(f"Connection limit reached, rejecting {client_addr}")
                client_sock.close()
                return

            client_sock.setblocking(False)
            # 设置 TCP Keepalive
            client_sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

            with self._lock:
                self._connections[fd] = (client_sock, client_addr, time.time())
                self._recv_buffers[fd] = bytearray()

            logger.info(f"New connection from {client_addr} (fd={fd})")

        except Exception as e:
            logger.error(f"Accept error: {e}")

    def _handle_client_data(self, fd, sock):
        """处理客户端数据，处理粘包和拆包"""
        try:
            data = sock.recv(4096)
            if not data:
                # 客户端关闭连接
                self._close_connection(fd)
                return

            with self._lock:
                self._recv_buffers[fd].extend(data)
                self._connections[fd] = (sock, self._connections[fd][1], time.time())

            # 尝试解析完整帧
            self._process_complete_frames(fd)

        except ConnectionResetError:
            self._close_connection(fd)
        except Exception as e:
            logger.error(f"Data handling error fd={fd}: {e}")
            self._close_connection(fd)

    def _process_complete_frames(self, fd):
        """从接收缓冲区中提取完整 Modbus TCP 帧"""
        while True:
            buf = self._recv_buffers[fd]
            if len(buf) < 7:
                break

            # 解析 MBAP 头
            trans_id, proto_id, length, unit_id = struct.unpack('>HHHB', buf[:7])

            if proto_id != 0:
                logger.warning(f"Invalid protocol ID {proto_id} from fd={fd}")
                self._close_connection(fd)
                return

            total_frame_len = 6 + length  # MBAP(6) + 后续字节(length)
            if len(buf) < total_frame_len:
                # 帧不完整，等待更多数据
                break

            # 提取完整帧
            frame = bytes(buf[:total_frame_len])
            del buf[:total_frame_len]

            # 处理帧
            response = self._handle_modbus_frame(frame)
            if response:
                try:
                    sock = self._connections[fd][0]
                    sock.send(response)
                except Exception as e:
                    logger.error(f"Send error fd={fd}: {e}")
                    self._close_connection(fd)
                    return

    def _handle_modbus_frame(self, frame):
        """解析 Modbus TCP 帧并生成响应"""
        if len(frame) < 8:
            return None

        trans_id, proto_id, length, unit_id = struct.unpack('>HHHB', frame[:7])
        pdu = frame[7:]

        if len(pdu) < 1:
            return None

        func_code = pdu[0]

        # 检查 Unit ID
        if unit_id not in self.register_maps:
            logger.warning(f"Unknown unit_id {unit_id}")
            return self._build_exception_response(trans_id, unit_id, func_code, ExceptionCode.SLAVE_DEVICE_FAILURE)

        reg_map = self.register_maps[unit_id]

        try:
            if func_code == FunctionCode.READ_HOLDING_REGISTERS:
                return self._handle_read_holding(trans_id, unit_id, pdu, reg_map)
            elif func_code == FunctionCode.READ_INPUT_REGISTERS:
                return self._handle_read_input(trans_id, unit_id, pdu, reg_map)
            elif func_code == FunctionCode.READ_COILS:
                return self._handle_read_coils(trans_id, unit_id, pdu, reg_map)
            elif func_code == FunctionCode.READ_DISCRETE_INPUTS:
                return self._handle_read_discrete(trans_id, unit_id, pdu, reg_map)
            elif func_code == FunctionCode.WRITE_SINGLE_REGISTER:
                return self._handle_write_single(trans_id, unit_id, pdu, reg_map)
            elif func_code == FunctionCode.WRITE_MULTIPLE_REGISTERS:
                return self._handle_write_multiple(trans_id, unit_id, pdu, reg_map)
            elif func_code == FunctionCode.WRITE_SINGLE_COIL:
                return self._handle_write_single_coil(trans_id, unit_id, pdu, reg_map)
            else:
                return self._build_exception_response(trans_id, unit_id, func_code, ExceptionCode.ILLEGAL_FUNCTION)

        except ValueError as e:
            if "Address out of range" in str(e):
                return self._build_exception_response(trans_id, unit_id, func_code, ExceptionCode.ILLEGAL_DATA_ADDRESS)
            elif "Illegal data" in str(e):
                return self._build_exception_response(trans_id, unit_id, func_code, ExceptionCode.ILLEGAL_DATA_VALUE)
            else:
                return self._build_exception_response(trans_id, unit_id, func_code, ExceptionCode.SLAVE_DEVICE_FAILURE)
        except Exception as e:
            logger.error(f"Frame handling error: {e}")
            return self._build_exception_response(trans_id, unit_id, func_code, ExceptionCode.SLAVE_DEVICE_FAILURE)

    def _handle_read_holding(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x03 读保持寄存器"""
        if len(pdu) < 5:
            raise ValueError("Illegal data value")
        start_addr = struct.unpack('>H', pdu[1:3])[0]
        count = struct.unpack('>H', pdu[3:5])[0]

        if count < 1 or count > 125:
            raise ValueError("Illegal data value")

        values = reg_map.read_holding_registers(start_addr, count)
        response_pdu = bytes([FunctionCode.READ_HOLDING_REGISTERS, len(values) * 2])
        for v in values:
            response_pdu += struct.pack('>H', v)

        return self._build_response(trans_id, unit_id, response_pdu)

    def _handle_read_input(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x04 读输入寄存器"""
        if len(pdu) < 5:
            raise ValueError("Illegal data value")
        start_addr = struct.unpack('>H', pdu[1:3])[0]
        count = struct.unpack('>H', pdu[3:5])[0]

        if count < 1 or count > 125:
            raise ValueError("Illegal data value")

        values = reg_map.read_input_registers(start_addr, count)
        response_pdu = bytes([FunctionCode.READ_INPUT_REGISTERS, len(values) * 2])
        for v in values:
            response_pdu += struct.pack('>H', v)

        return self._build_response(trans_id, unit_id, response_pdu)

    def _handle_read_coils(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x01 读线圈"""
        if len(pdu) < 5:
            raise ValueError("Illegal data value")
        start_addr = struct.unpack('>H', pdu[1:3])[0]
        count = struct.unpack('>H', pdu[3:5])[0]

        if count < 1 or count > 2000:
            raise ValueError("Illegal data value")

        coils = reg_map.read_coils(start_addr, count)
        # 打包为字节
        byte_count = (count + 7) // 8
        coil_bytes = bytearray(byte_count)
        for i, val in enumerate(coils):
            if val:
                coil_bytes[i // 8] |= (1 << (i % 8))

        response_pdu = bytes([FunctionCode.READ_COILS, byte_count]) + bytes(coil_bytes)
        return self._build_response(trans_id, unit_id, response_pdu)

    def _handle_read_discrete(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x02 读离散输入"""
        if len(pdu) < 5:
            raise ValueError("Illegal data value")
        start_addr = struct.unpack('>H', pdu[1:3])[0]
        count = struct.unpack('>H', pdu[3:5])[0]

        if count < 1 or count > 2000:
            raise ValueError("Illegal data value")

        # 离散输入和线圈共用存储结构
        inputs = reg_map.read_coils(start_addr, count)  # 复用方法
        byte_count = (count + 7) // 8
        input_bytes = bytearray(byte_count)
        for i, val in enumerate(inputs):
            if val:
                input_bytes[i // 8] |= (1 << (i % 8))

        response_pdu = bytes([FunctionCode.READ_DISCRETE_INPUTS, byte_count]) + bytes(input_bytes)
        return self._build_response(trans_id, unit_id, response_pdu)

    def _handle_write_single(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x06 写单个寄存器"""
        if len(pdu) < 5:
            raise ValueError("Illegal data value")
        addr = struct.unpack('>H', pdu[1:3])[0]
        value = struct.unpack('>H', pdu[3:5])[0]

        reg_map.write_holding_register(addr, value)

        # 响应：回显请求
        response_pdu = bytes([FunctionCode.WRITE_SINGLE_REGISTER]) + pdu[1:5]
        return self._build_response(trans_id, unit_id, response_pdu)

    def _handle_write_multiple(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x10 写多个寄存器"""
        if len(pdu) < 7:
            raise ValueError("Illegal data value")
        start_addr = struct.unpack('>H', pdu[1:3])[0]
        count = struct.unpack('>H', pdu[3:5])[0]
        byte_count = pdu[5]

        if count < 1 or count > 123 or byte_count != count * 2:
            raise ValueError("Illegal data value")

        values = []
        for i in range(count):
            offset = 6 + i * 2
            val = struct.unpack('>H', pdu[offset:offset+2])[0]
            values.append(val)

        reg_map.write_holding_registers(start_addr, values)

        # 响应：起始地址 + 写入数量
        response_pdu = bytes([FunctionCode.WRITE_MULTIPLE_REGISTERS]) + struct.pack('>HH', start_addr, count)
        return self._build_response(trans_id, unit_id, response_pdu)

    def _handle_write_single_coil(self, trans_id, unit_id, pdu, reg_map):
        """处理 0x05 写单个线圈"""
        if len(pdu) < 5:
            raise ValueError("Illegal data value")
        addr = struct.unpack('>H', pdu[1:3])[0]
        coil_value = struct.unpack('>H', pdu[3:5])[0]

        if coil_value not in (0x0000, 0xFF00):
            raise ValueError("Illegal data value")

        reg_map.write_coil(addr, bool(coil_value == 0xFF00))

        response_pdu = bytes([FunctionCode.WRITE_SINGLE_COIL]) + pdu[1:5]
        return self._build_response(trans_id, unit_id, response_pdu)

    def _build_response(self, trans_id, unit_id, response_pdu):
        """构造 Modbus TCP 响应帧"""
        length = len(response_pdu) + 1  # +1 for unit_id
        header = struct.pack('>HHHB', trans_id, 0, length, unit_id)
        return header + response_pdu

    def _build_exception_response(self, trans_id, unit_id, func_code, exception_code):
        """构造异常响应帧"""
        response_pdu = bytes([func_code | 0x80, exception_code])
        return self._build_response(trans_id, unit_id, response_pdu)

    def _close_connection(self, fd):
        """关闭指定连接"""
        with self._lock:
            if fd in self._connections:
                sock, addr, _ = self._connections[fd]
                try:
                    sock.close()
                except:
                    pass
                del self._connections[fd]
                if fd in self._recv_buffers:
                    del self._recv_buffers[fd]
                logger.info(f"Connection closed: {addr} (fd={fd})")

    def _cleanup_stale_connections(self, timeout=300):
        """清理超时连接"""
        now = time.time()
        stale_fds = []
        with self._lock:
            for fd, (sock, addr, last_active) in self._connections.items():
                if now - last_active > timeout:
                    stale_fds.append(fd)

        for fd in stale_fds:
            logger.info(f"Cleaning up stale connection fd={fd}")
            self._close_connection(fd)

# server_main.py
from modbus_tcp_server_select import ModbusTCPServer
from modbus_server_core import RegisterMap
import threading
import time
import random

def simulate_sensor_data(reg_map, stop_event):
    """模拟传感器数据更新（输入寄存器）"""
    while not stop_event.is_set():
        # 模拟温度 20~30℃，湿度 40~60%RH
        temp = int((20 + random.uniform(0, 10)) * 10)      # 0x0000: 温度 ×10
        hum = int((40 + random.uniform(0, 20)) * 10)       # 0x0001: 湿度 ×10
        dew = int((15 + random.uniform(0, 8)) * 10)        # 0x0002: 露点 ×10
        status = 0x0001                                    # 0x0003: 状态（正常）

        reg_map.update_input_register(0x0000, temp)
        reg_map.update_input_register(0x0001, hum)
        reg_map.update_input_register(0x0002, dew)
        reg_map.update_input_register(0x0003, status)

        # 模拟线圈状态（设备运行/报警）
        reg_map.write_coil(0, True)     # 设备运行
        reg_map.write_coil(1, False)    # 无报警

        time.sleep(1)

def main():
    # 创建寄存器映射
    reg_map_1 = RegisterMap(unit_id=1)
    reg_map_2 = RegisterMap(unit_id=2)

    # 初始化保持寄存器（配置参数）
    reg_map_1.write_holding_register(0x0000, 250)   # 温度上限 25.0℃
    reg_map_1.write_holding_register(0x0001, 180)   # 温度下限 18.0℃
    reg_map_1.write_holding_register(0x0002, 600)   # 湿度上限 60.0%
    reg_map_1.write_holding_register(0x0003, 450)   # 湿度下限 45.0%

    reg_map_2.write_holding_register(0x0000, 240)
    reg_map_2.write_holding_register(0x0001, 180)
    reg_map_2.write_holding_register(0x0002, 550)
    reg_map_2.write_holding_register(0x0003, 450)

    # 创建服务器
    server = ModbusTCPServer(host='0.0.0.0', port=502, max_connections=50)
    server.add_register_map(1, reg_map_1)
    server.add_register_map(2, reg_map_2)

    # 启动数据模拟线程
    stop_event = threading.Event()
    sim_thread = threading.Thread(target=simulate_sensor_data, args=(reg_map_1, stop_event), daemon=True)
    sim_thread.start()

    # 启动服务器
    try:
        server.start()
    except KeyboardInterrupt:
        print("\nShutting down...")
        stop_event.set()
        server.stop()

if __name__ == '__main__':
    main()

# modbus_tcp_server_epoll.py
import select
import socket
import threading
import time
import struct
from modbus_tcp_server_select import ModbusTCPServer

class EpollModbusTCPServer(ModbusTCPServer):
    """基于 epoll 的高性能 Modbus TCP 服务端"""

    def _event_loop(self):
        """epoll 事件循环"""
        epoll_fd = select.epoll()
        epoll_fd.register(self._server_sock.fileno(), select.EPOLLIN | select.EPOLLET)

        # fd → socket 映射
        fd_to_sock = {self._server_sock.fileno(): self._server_sock}
        fd_to_addr = {}

        try:
            while self._running:
                events = epoll_fd.poll(timeout=1.0)

                for fd, event in events:
                    if fd == self._server_sock.fileno():
                        # 新连接
                        self._accept_new_connection_epoll(epoll_fd, fd_to_sock, fd_to_addr)
                    elif event & select.EPOLLIN:
                        # 可读
                        try:
                            self._handle_client_data(fd, fd_to_sock[fd])
                        except Exception as e:
                            logger.error(f"epoll read error fd={fd}: {e}")
                            self._close_connection_epoll(fd, epoll_fd, fd_to_sock, fd_to_addr)
                    elif event & (select.EPOLLHUP | select.EPOLLERR):
                        # 连接错误/挂断
                        logger.warning(f"epoll error/hup fd={fd}")
                        self._close_connection_epoll(fd, epoll_fd, fd_to_sock, fd_to_addr)

                # 清理超时连接
                self._cleanup_stale_connections()

        finally:
            epoll_fd.close()

    def _accept_new_connection_epoll(self, epoll_fd, fd_to_sock, fd_to_addr):
        """epoll 模式接受新连接"""
        while True:
            try:
                client_sock, client_addr = self._server_sock.accept()
            except socket.error:
                break  # 没有更多连接

            fd = client_sock.fileno()
            if len(self._connections) >= self.max_connections:
                client_sock.close()
                continue

            client_sock.setblocking(False)
            client_sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

            epoll_fd.register(fd, select.EPOLLIN | select.EPOLLET)
            fd_to_sock[fd] = client_sock
            fd_to_addr[fd] = client_addr

            with self._lock:
                self._connections[fd] = (client_sock, client_addr, time.time())
                self._recv_buffers[fd] = bytearray()

            logger.info(f"epoll: new connection {client_addr} fd={fd}")

    def _close_connection_epoll(self, fd, epoll_fd, fd_to_sock, fd_to_addr):
        """epoll 模式关闭连接"""
        epoll_fd.unregister(fd)
        if fd in fd_to_sock:
            fd_to_sock[fd].close()
            del fd_to_sock[fd]
        if fd in fd_to_addr:
            del fd_to_addr[fd]
        self._close_connection(fd)

# modbus_tcp_server_threaded.py
import socket
import threading
import queue
import time
from modbus_server_core import RegisterMap

class ThreadedModbusTCPServer:
    """每个连接一个线程的 Modbus TCP 服务端"""

    def __init__(self, host='0.0.0.0', port=502, max_threads=100):
        self.host = host
        self.port = port
        self.max_threads = max_threads
        self.register_maps = {}
        self._running = False
        self._server_sock = None
        self._threads = {}
        self._thread_lock = threading.Lock()

    def add_register_map(self, unit_id, reg_map):
        self.register_maps[unit_id] = reg_map

    def start(self):
        self._server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._server_sock.bind((self.host, self.port))
        self._server_sock.listen(128)
        self._running = True

        logger.info(f"Threaded Modbus TCP Server on {self.host}:{self.port}")

        while self._running:
            try:
                client_sock, client_addr = self._server_sock.accept()
                with self._thread_lock:
                    if len(self._threads) >= self.max_threads:
                        logger.warning(f"Thread limit reached, rejecting {client_addr}")
                        client_sock.close()
                        continue

                t = threading.Thread(
                    target=self._handle_connection,
                    args=(client_sock, client_addr),
                    daemon=True
                )
                t.start()
                self._threads[t.ident] = t

            except Exception as e:
                logger.error(f"Accept error: {e}")

    def _handle_connection(self, sock, addr):
        """处理单个客户端连接"""
        logger.info(f"Thread {threading.current_thread().ident}: handling {addr}")
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        recv_buffer = bytearray()
        last_active = time.time()

        try:
            while self._running:
                sock.settimeout(30.0)
                try:
                    data = sock.recv(4096)
                except socket.timeout:
                    # 检查是否超时
                    if time.time() - last_active > 300:
                        logger.info(f"Connection {addr} timed out")
                        break
                    continue

                if not data:
                    break

                last_active = time.time()
                recv_buffer.extend(data)

                # 处理完整帧
                while len(recv_buffer) >= 7:
                    trans_id, proto_id, length, unit_id = struct.unpack('>HHHB', recv_buffer[:7])
                    total_len = 6 + length
                    if len(recv_buffer) < total_len:
                        break

                    frame = bytes(recv_buffer[:total_len])
                    del recv_buffer[:total_len]

                    # 解析并响应
                    response = self._handle_frame(frame)
                    if response:
                        sock.send(response)

        except Exception as e:
            logger.error(f"Connection handler error {addr}: {e}")
        finally:
            sock.close()
            logger.info(f"Connection closed: {addr}")
            with self._thread_lock:
                self._threads.pop(threading.current_thread().ident, None)

    def _handle_frame(self, frame):
        """解析 Modbus TCP 帧（简化版，完整版同上）"""
        # 复用 select 版本的逻辑
        pass

    def stop(self):
        self._running = False
        if self._server_sock:
            self._server_sock.close()

场景 1：两个客户端同时写同一寄存器
  客户端 A: 写 40001 = 250
  客户端 B: 写 40001 = 300
  结果：取决于线程调度，最终值不确定

场景 2：客户端读的同时模拟线程写输入寄存器
  读线程: 读 30001~30004
  写线程: 更新 30002
  结果：可能读到一半新值一半旧值

场景 3：连接管理并发修改
  主线程: 清理超时连接，删除 fd
  epoll: 返回该 fd 的事件
  结果：Use-after-free

# rwlock.py
import threading

class ReadWriteLock:
    """读写锁，读多写少场景性能优于互斥锁"""
    def __init__(self):
        self._read_ready = threading.Condition(threading.Lock())
        self._readers = 0

    def acquire_read(self):
        self._read_ready.acquire()
        try:
            self._readers += 1
        finally:
            self._read_ready.release()

    def release_read(self):
        self._read_ready.acquire()
        try:
            self._readers -= 1
            if self._readers == 0:
                self._read_ready.notify_all()
        finally:
            self._read_ready.release()

    def acquire_write(self):
        self._read_ready.acquire()
        while self._readers > 0:
            self._read_ready.wait()

    def release_write(self):
        self._read_ready.release()

# 使用上下文管理器
class ReadLock:
    def __init__(self, rwlock):
        self.rwlock = rwlock
    def __enter__(self):
        self.rwlock.acquire_read()
        return self
    def __exit__(self, *args):
        self.rwlock.release_read()

class WriteLock:
    def __init__(self, rwlock):
        self.rwlock = rwlock
    def __enter__(self):
        self.rwlock.acquire_write()
        return self
    def __exit__(self, *args):
        self.rwlock.release_write()

# 查看当前限制
ulimit -n

# 临时调整
ulimit -n 65535

# 永久调整（/etc/security/limits.conf）
* soft nofile 65535
* hard nofile 65535

# systemd 服务文件
[Service]
LimitNOFILE=65535

# /etc/sysctl.conf
net.core.somaxconn = 4096        # listen backlog
net.core.netdev_max_backlog = 5000
net.ipv4.tcp_max_syn_backlog = 4096
net.ipv4.tcp_keepalive_time = 600
net.ipv4.tcp_keepalive_intvl = 60
net.ipv4.tcp_keepalive_probes = 3
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 30
net.ipv4.ip_local_port_range = 1024 65535

sysctl -p

# 1. 使用 bytes 而非 bytearray 做频繁拼接
# 2. struct 预编译格式字符串
import struct
_pack_hh = struct.Struct('>HH')
_pack_hhhb = struct.Struct('>HHHB')

# 3. 使用 __slots__ 减少对象内存
class ConnectionInfo:
    __slots__ = ['sock', 'addr', 'last_active', 'buffer']
    def __init__(self, sock, addr):
        self.sock = sock
        self.addr = addr
        self.last_active = time.time()
        self.buffer = bytearray()

# 4. 使用 memoryview 避免拷贝
buf = bytearray(4096)
view = memoryview(buf)

# test_server.py
import socket
import struct
import threading
import time

def test_client(server_host, server_port, unit_id=1):
    """测试客户端"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((server_host, server_port))
    sock.settimeout(5.0)

    trans_id = 0

    # 测试 1: 读输入寄存器
    trans_id += 1
    mbap = struct.pack('>HHHB', trans_id, 0, 6, unit_id)
    pdu = struct.pack('>BHH', 0x04, 0, 4)    # 读 4 个输入寄存器
    sock.send(mbap + pdu)

    response = sock.recv(1024)
    print(f"Test 1 - Read Input Registers: {len(response)} bytes")
    if len(response) >= 9:
        _, _, length, resp_unit = struct.unpack('>HHHB', response[:7])
        func_code = response[7]
        print(f"  Function: 0x{func_code:02X}, Length: {length}")

    # 测试 2: 写单个寄存器
    trans_id += 1
    mbap = struct.pack('>HHHB', trans_id, 0, 6, unit_id)
    pdu = struct.pack('>BHH', 0x06, 0x0000, 999)   # 写 40001 = 999
    sock.send(mbap + pdu)

    response = sock.recv(1024)
    print(f"Test 2 - Write Single Register: {len(response)} bytes")

    # 测试 3: 读保持寄存器验证写入
    trans_id += 1
    mbap = struct.pack('>HHHB', trans_id, 0, 6, unit_id)
    pdu = struct.pack('>BHH', 0x03, 0x0000, 1)    # 读 40001
    sock.send(mbap + pdu)

    response = sock.recv(1024)
    print(f"Test 3 - Read Back: {len(response)} bytes")
    if len(response) >= 11:
        value = struct.unpack('>H', response[9:11])[0]
        print(f"  Value: {value} (expected 999)")

    # 测试 4: 非法功能码
    trans_id += 1
    mbap = struct.pack('>HHHB', trans_id, 0, 6, unit_id)
    pdu = struct.pack('>BHH', 0xFF, 0, 0)    # 非法功能码
    sock.send(mbap + pdu)

    response = sock.recv(1024)
    print(f"Test 4 - Illegal Function: {len(response)} bytes")
    if len(response) >= 8:
        resp_func = response[7]
        if resp_func & 0x80:
            exc_code = response[8]
            print(f"  Exception: 0x{exc_code:02X}")

    sock.close()
    print("All tests completed")

if __name__ == '__main__':
    test_client('127.0.0.1', 502, unit_id=1)

# stress_test.py
import socket
import struct
import threading
import time
import random

class StressClient:
    def __init__(self, server_host, server_port, client_id):
        self.host = server_host
        self.port = server_port
        self.client_id = client_id
        self.requests = 0
        self.errors = 0

    def run(self, duration=60):
        """运行压力测试"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((self.host, self.port))
        sock.settimeout(2.0)

        start = time.time()
        trans_id = 0

        while time.time() - start < duration:
            try:
                trans_id += 1
                mbap = struct.pack('>HHHB', trans_id, 0, 6, 1)
                # 随机读 1~4 个输入寄存器
                count = random.randint(1, 4)
                start_addr = random.randint(0, 10)
                pdu = struct.pack('>BHH', 0x04, start_addr, count)
                sock.send(mbap + pdu)

                response = sock.recv(1024)
                if len(response) < 9:
                    self.errors += 1
                else:
                    self.requests += 1

            except Exception:
                self.errors += 1
                try:
                    sock.close()
                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    sock.connect((self.host, self.port))
                    sock.settimeout(2.0)
                except:
                    break

        sock.close()

def stress_test(server_host='127.0.0.1', server_port=502, clients=50, duration=60):
    """多客户端并发压力测试"""
    threads = []
    clients_list = []

    start = time.time()

    for i in range(clients):
        c = StressClient(server_host, server_port, i)
        clients_list.append(c)
        t = threading.Thread(target=c.run, args=(duration,))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

    elapsed = time.time() - start
    total_requests = sum(c.requests for c in clients_list)
    total_errors = sum(c.errors for c in clients_list)

    print(f"\n{'='*50}")
    print(f"压力测试结果")
    print(f"{'='*50}")
    print(f"客户端数: {clients}")
    print(f"持续时间: {elapsed:.1f}s")
    print(f"总请求数: {total_requests}")
    print(f"总错误数: {total_errors}")
    print(f"QPS: {total_requests/elapsed:.1f}")
    print(f"错误率: {total_errors/(total_requests+total_errors)*100:.2f}%" if (total_requests+total_errors) > 0 else "N/A")

if __name__ == '__main__':
    stress_test(clients=50, duration=60)

class ConnectionLimiter:
    def __init__(self, max_per_ip=5, max_total=200):
        self.max_per_ip = max_per_ip
        self.max_total = max_total
        self._ip_count = defaultdict(int)
        self._lock = threading.Lock()

    def check(self, client_addr):
        ip = client_addr[0]
        with self._lock:
            total = sum(self._ip_count.values())
            if total >= self.max_total:
                return False, "Total connection limit exceeded"
            if self._ip_count[ip] >= self._max_per_ip:
                return False, f"Per-IP limit exceeded for {ip}"
            self._ip_count[ip] += 1
            return True, "OK"

    def release(self, client_addr):
        ip = client_addr[0]
        with self._lock:
            self._ip_count[ip] = max(0, self._ip_count[ip] - 1)
            if self._ip_count[ip] == 0:
                del self._ip_count[ip]

class AuditLogger:
    def __init__(self, log_file="modbus_audit.log"):
        self.log_file = log_file

    def log_request(self, client_addr, unit_id, func_code, start_addr, count, result="OK"):
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        with open(self.log_file, 'a') as f:
            f.write(f"{timestamp} | {client_addr[0]}:{client_addr[1]} | "
                    f"Unit={unit_id} | Func=0x{func_code:02X} | "
                    f"Addr=0x{start_addr:04X} | Count={count} | {result}\n")

def graceful_shutdown(server, signum, frame):
    """信号处理函数"""
    logger.info(f"Received signal {signum}, shutting down gracefully...")
    server.stop()
    # 等待所有连接关闭
    time.sleep(2)
    logger.info("Server stopped")
    exit(0)

import signal
signal.signal(signal.SIGTERM, lambda s, f: graceful_shutdown(server, s, f))
signal.signal(signal.SIGINT, lambda s, f: graceful_shutdown(server, s, f))