# -*- coding: utf-8 -*- """ test_ModbusWriteCache_simulation.py Modbus TCP 写缓存队列（FC_ModbusWrite_PutTask / FC_ModbusWrite_CacheRun）仿真验证 怎么用： "C:/Users/HawkZhang/.workbuddy/binaries/python/versions/3.13.12/python.exe" test_ModbusWriteCache_simulation.py 为什么要有这个脚本： 写队列这种东西"看着对"完全不作数 —— 它的缺陷只在极窄的时序窗口里露头 （对端卡死、请求未被受理、重试耗尽、队列刚好出队那一拍…），现场是"偶发、 说不清"。这里用 Python 1:1 复刻两份状态机的离散递推，把窗口造出来， 再拿"新版 vs 原稿"跑同样的激励做对照，看差异到底出在哪。 仿真里对 MB_CLIENT 的行为按西门子 S7-1200 系统手册的原文建模： · REQ=1 且当前没有作业在跑 -> 起动一个作业 · DONE / ERROR **只维持一个扫描周期**，STATUS 也只在那一拍有效 · 作业进行中再次以 REQ=1 调用，不会发出新报文； 但"当前请求一完成、若 REQ 仍为 1，就会再处理一次" · DISCONNECT=1 且 REQ=0 -> 断开，STATUS=16#7003 · 16#7004 的官方定义是"连接已建立且处于受监视状态、未激活任何作业处理"， 也就是**正常空闲态**，不是故障 —— 这一条原稿读反了 "REQ 保持高电平会不会自动重发"手册措辞有歧义，所以模型里做成开关 REQ_LEVEL_RESEND，两种解释都跑一遍，结论分开报。 时间基准：1 个扫描周期 = 10 ms（OB1 循环量级）。 """ import sys SCAN_MS = 10 QUEUE_LEN = 100 CAPACITY = 99 PASS, FAIL = [], [] def check(name, cond, detail=''): (PASS if cond else FAIL).append(name) print(' [%s] %-56s %s' % ('OK' if cond else '!!', name, detail)) def scans(ms): return max(1, int(round(ms / SCAN_MS))) # ---------------------------------------------------------------------------- # 定时器（TON），按扫描周期计 # ---------------------------------------------------------------------------- class Ton: def __init__(self): self.et = 0 self.q = False def __call__(self, in_, pt): if not in_: self.et = 0 self.q = False elif self.et < pt: self.et += 1 self.q = (self.et >= pt) return self.q # ---------------------------------------------------------------------------- # MB_CLIENT 搭档模型 # ---------------------------------------------------------------------------- class MbClient: """mode: 'ok' 正常：受理后 resp_scans 拍返回 DONE 'err' 受理后返回 ERROR + status_code 'dead_busy' 受理了但永远不返回 DONE/ERROR（对端 TCP 半开） 'no_accept' 根本不受理：REQ=1 也不拉 BUSY、不报错（端口被别的请求占用） """ def __init__(self, mode='ok', resp_scans=5, status_code=0x80C8, req_level_resend=True): self.mode = mode self.resp_scans = resp_scans self.status_code = status_code self.req_level_resend = req_level_resend self.done = False self.error = False self.busy = False self.status = 0x7000 self.connected = False self.job = None self.remain = 0 self.frames = [] # 真正发出去的报文 —— "到底写了什么"的铁证 self.disconnects = 0 def call(self, req, disconnect, unit_id, addr, length, value): # 结果脉冲只活一个扫描周期：本次调用先把上一拍的清掉 self.done = False self.error = False if disconnect: if self.connected: self.connected = False self.disconnects += 1 self.busy = False self.job = None self.status = 0x7003 self.done = True else: self.status = 0x7000 return if not self.connected: if self.mode == 'no_accept': self.status = 0x8200 return self.connected = True if self.job is None: if not req: self.status = 0x7004 if self.connected else 0x7000 return if self.mode == 'no_accept': # 端口正在处理另一个请求：不受理、也不报错，REQ 白置 self.status = 0x8200 return self._start(unit_id, addr, length, value) return # 作业进行中 self.busy = True self.status = 0x7005 if self.mode == 'dead_busy': return # 永远不结账 self.remain -= 1 if self.remain > 0: return self.busy = False self.job = None if self.mode == 'err': self.error = True self.status = self.status_code else: self.done = True self.status = 0x7004 # 已连接、无作业处理中 if self.req_level_resend and req: # 手册原文：当前请求一完成，"若以 REQ=1 执行，就可以处理新请求" self._start(unit_id, addr, length, value) def _start(self, unit_id, addr, length, value): self.job = (unit_id, addr, length, value) self.remain = self.resp_scans self.busy = True self.status = 0x7001 self.frames.append((unit_id, addr, value)) # ---------------------------------------------------------------------------- # 新版：DB / PutTask / CacheRun # ---------------------------------------------------------------------------- def new_db(): return dict( QUEUE_MAX_LEN=QUEUE_LEN, QUEUE_CAPACITY=CAPACITY, RETRY_MAX=3, ENABLE_WRITE=True, RETRY_ON_PROTO_ERR=False, ADDR_OFFSET=0, TRIG_MODE=0, # 自 v2.1 起块内已忽略此字段，留着只为对齐真实 DB 结构 T_ACCEPT=scans(1000), T_REQ=scans(6000), T_DISC=scans(300), T_RECONN=scans(500), q=[None] * QUEUE_LEN, wr=0, rd=0, empty=True, full=False, cnt=0, state=0, cur=None, busy=False, err=False, exec=False, disc=False, last_trig=False, call_cnt=0, write_locked=False, mb_unit=0, mb_addr=0, wdata=0, VER=0x20, nonce=0, ok=0, retry=0, err_cnt=0, drop=0, reject=0, timeout=0, addrwarn=0, status_live=0x7000, status_lat=0, last_err_status=0, last_err_task=None, tm_accept=Ton(), tm_req=Ton(), tm_disc=Ton(), tm_reconn=Ton(), ) def new_put_task(db, slave, addr, value, trig): out = dict(AddOk=False, Merged=False, QueueFullFlag=False, Rejected=False, RejectCode=0, AddrWarn=False, MultiCall=False, TrigHoldWarn=False) # 1 指针自愈 + 水位（判满直接看条目数，不看上一拍的标记） if not (0 <= db['wr'] <= db['QUEUE_MAX_LEN'] - 1): db['wr'] = 0 if not (0 <= db['rd'] <= db['QUEUE_MAX_LEN'] - 1): db['rd'] = 0 out['QueueFullFlag'] = (db['cnt'] >= db['QUEUE_CAPACITY']) # 本周期第几次调用。v2.1：多次调用是正常用法，这个计数只用于两件事： # 认出"本周期第一次调用"（做跨周期比较）和 MultiCall 诊断 db['call_cnt'] += 1 if db['call_cnt'] > 30000: db['call_cnt'] = 1 # CacheRun 没在跑时 Int 会溢出，先自保 first_in_scan = (db['call_cnt'] == 1) out['MultiCall'] = not first_in_scan # 2 触发（v2.1：块内不做边沿检测，AddTrigger 本身就是单周期脉冲） b_edge = bool(trig) # 保险：只在每周期第一次调用时跟"上周期最后一次调用的值"比一次。 # 连续两拍为 TRUE = 触发位没取沿。只提示，不拦截 if first_in_scan and trig and db['last_trig']: out['TrigHoldWarn'] = True db['last_trig'] = bool(trig) # 3 地址合成 + 参数体检 dw_addr = addr + db['ADDR_OFFSET'] if b_edge: if slave == 0: out['Rejected'] = True out['RejectCode'] = 1 db['reject'] += 1 if dw_addr > 65535 and not out['Rejected']: out['Rejected'] = True out['RejectCode'] = 2 db['reject'] += 1 if dw_addr < 40001 and not out['Rejected']: out['AddrWarn'] = True db['addrwarn'] += 1 # 4 入队 if b_edge and not out['Rejected']: if out['QueueFullFlag']: db['drop'] += 1 else: i_cnt = db['cnt'] i_slot = db['rd'] if db['busy']: i_slot = (i_slot + 1) % db['QUEUE_MAX_LEN'] i_cnt -= 1 found = False for _ in range(max(0, i_cnt)): e = db['q'][i_slot] if e and e['valid'] and e['slave'] == slave and e['addr'] == dw_addr: e['value'] = value e['retry'] = db['RETRY_MAX'] found = True break i_slot = (i_slot + 1) % db['QUEUE_MAX_LEN'] if found: out['Merged'] = True out['AddOk'] = True else: db['nonce'] += 1 db['q'][db['wr']] = dict(slave=slave, addr=dw_addr, value=value, retry=db['RETRY_MAX'], valid=True, seq=db['nonce']) db['wr'] = (db['wr'] + 1) % db['QUEUE_MAX_LEN'] db['cnt'] += 1 db['empty'] = False db['full'] = (db['cnt'] >= db['QUEUE_CAPACITY']) out['AddOk'] = True return out def new_cache_run(db, mb): """严格按 SCL 的顺序：水位 -> 状态机 -> 出队收尾 -> 诊断 -> 调用 MB_CLIENT""" out = dict(Busy=False, Err=False, Done=False, Dropped=False) done_p = dropped_p = False i_pop = 0 # 1 指针自愈 + 水位（从两个指针现算，不依赖谁先执行） if not (0 <= db['rd'] <= db['QUEUE_MAX_LEN'] - 1): db['rd'] = 0 l = db['wr'] - db['rd'] if l < 0: l += db['QUEUE_MAX_LEN'] if l < 0 or l > db['QUEUE_CAPACITY']: l = 0 db['cnt'], db['empty'] = l, (l == 0) db['full'] = (l >= db['QUEUE_CAPACITY']) # 2 定时器 db['tm_accept'](db['state'] == 10, db['T_ACCEPT']) db['tm_req'](db['state'] == 20, db['T_REQ']) db['tm_disc'](db['state'] == 60, db['T_DISC']) db['tm_reconn'](db['state'] == 65, db['T_RECONN']) # 3 状态机 st = db['state'] if st == 0: db['exec'] = False db['disc'] = False if not db['empty']: db['cur'] = dict(db['q'][db['rd']]) if db['cur']['valid']: if db['ENABLE_WRITE']: db['busy'] = True db['state'] = 10 else: i_pop = 1 else: db['busy'] = False elif st == 10: # SCL：先写作业参数，再把 REQ 置起来，再读输出 db['mb_unit'] = db['cur']['slave'] db['mb_addr'] = db['cur']['addr'] db['wdata'] = db['cur']['value'] db['exec'] = True if mb.done: db['status_lat'] = 0 db['busy'] = False db['err'] = False i_pop = 2 db['state'] = 0 elif mb.error: db['status_lat'] = mb.status db['state'] = 30 elif mb.busy: db['state'] = 20 elif db['tm_accept'].q: db['status_lat'] = 0xFFFE db['timeout'] += 1 db['state'] = 30 elif st == 20: db['exec'] = False if mb.done: db['status_lat'] = 0 db['busy'] = False db['err'] = False i_pop = 2 db['state'] = 0 elif mb.error: db['status_lat'] = mb.status db['state'] = 30 elif db['tm_req'].q: db['status_lat'] = 0xFFFF db['timeout'] += 1 db['state'] = 30 elif st == 30: db['exec'] = False w = db['status_lat'] b_no_retry = (0x8180 <= w <= 0x818F) or (0x8380 <= w <= 0x838F) if 0x8380 <= w <= 0x838F and db['RETRY_ON_PROTO_ERR']: b_no_retry = False b_reconn = (w >= 0xFF00) or (0x80C0 <= w <= 0x80FF) if b_no_retry: db['busy'] = False db['err'] = True db['last_err_status'] = db['status_lat'] db['last_err_task'] = dict(db['cur']) i_pop = 3 db['state'] = 0 else: db['cur']['retry'] -= 1 if db['cur']['retry'] > 0: db['retry'] += 1 db['state'] = 60 if b_reconn else 10 else: db['busy'] = False db['err'] = True db['last_err_status'] = db['status_lat'] db['last_err_task'] = dict(db['cur']) i_pop = 3 db['state'] = 0 elif st == 60: db['exec'] = False db['disc'] = True if db['tm_disc'].q: db['state'] = 65 elif st == 65: db['exec'] = False db['disc'] = False if db['tm_reconn'].q: db['state'] = 10 # 4 出队收尾 if i_pop > 0: db['q'][db['rd']]['valid'] = False db['rd'] = (db['rd'] + 1) % db['QUEUE_MAX_LEN'] if db['cnt'] > 0: db['cnt'] -= 1 db['empty'] = (db['cnt'] == 0) db['full'] = (db['cnt'] >= db['QUEUE_CAPACITY']) if i_pop == 2: db['ok'] += 1 done_p = True elif i_pop == 3: db['err_cnt'] += 1 dropped_p = True # 5 稳态诊断 db['write_locked'] = (not db['empty']) and (not db['ENABLE_WRITE']) db['VER'] = 0x20 # 6 调用 MB_CLIENT mb.call(db['exec'], db['disc'], db['mb_unit'], db['mb_addr'], 1, db['wdata']) db['status_live'] = mb.status db['call_cnt'] = 0 out['Busy'], out['Err'] = db['busy'], db['err'] out['Done'], out['Dropped'] = done_p, dropped_p return out class NewSim: def __init__(self, mb, **cfg): self.db = new_db() self.db.update(cfg) self.mb = mb def scan(self, puts=()): res = new_cache_run(self.db, self.mb) self.last_outs = [] if puts: for (slave, addr, val, trig) in puts: self.last_outs.append(new_put_task(self.db, slave, addr, val, trig)) else: # 真实 PLC 里 PutTask 是每周期都被调用的，只是触发信号为 FALSE。 # 少了这一拍，"上周期最后一次调用的值"就跟不上，跨周期比较会失真。 self.last_outs.append(new_put_task(self.db, 1, 40001, 0, False)) return res # ---------------------------------------------------------------------------- # 原稿（v0.1） # ---------------------------------------------------------------------------- def old_db(): return dict(q=[None] * QUEUE_LEN, wr=0, rd=0, empty=True, full=False, state=0, cur=None, busy=False, err=False, exec=False, mb_unit=0, mb_addr=0, wdata=0, retry_max=3, sent_ok=0, sent_fail=0, resend_hits=0) def old_put_task(db, slave, addr, value, trig): """原稿：注释写"上升沿添加任务"，代码却是电平触发""" if trig and not db['full']: db['q'][db['wr']] = dict(slave=slave, addr=addr, value=value, retry=db['retry_max'], valid=True) db['wr'] = (db['wr'] + 1) % 100 return True return False def old_cache_run(db, mb): """原稿：无兜底超时、重试一刀切、无统计、无锁存""" l = db['wr'] - db['rd'] if l < 0: l += 100 db['empty'] = (l == 0) db['full'] = (l >= 99) st = db['state'] if st == 0: db['exec'] = False if not db['empty']: db['cur'] = dict(db['q'][db['rd']]) if db['cur']['valid']: db['state'] = 1 db['busy'] = True else: db['rd'] = (db['rd'] + 1) % 100 else: db['busy'] = False elif st == 1: db['mb_unit'] = db['cur']['slave'] db['mb_addr'] = db['cur']['addr'] db['wdata'] = db['cur']['value'] db['exec'] = True db['state'] = 2 elif st == 2: db['exec'] = False if mb.done: db['q'][db['rd']]['valid'] = False db['rd'] = (db['rd'] + 1) % 100 db['err'] = False db['state'] = 0 db['sent_ok'] += 1 elif mb.error: db['cur']['retry'] -= 1 if db['cur']['retry'] > 0: db['state'] = 1 else: db['q'][db['rd']]['valid'] = False db['rd'] = (db['rd'] + 1) % 100 db['err'] = True db['state'] = 0 db['sent_fail'] += 1 # ==== 原稿的"程序卡住重新下发"分支（16#7004 是正常空闲态） ==== if mb.status == 0x7004 and db['busy'] and db['state'] == 2 and not db['exec']: db['exec'] = True db['resend_hits'] += 1 mb.call(db['exec'], 0, db['mb_unit'], db['mb_addr'], 1, db['wdata']) class OldSim: def __init__(self, mb): self.db = old_db() self.mb = mb def scan(self, puts=()): old_cache_run(self.db, self.mb) for (slave, addr, val, trig) in puts: old_put_task(self.db, slave, addr, val, trig) return self.db def run(sim, n, puts_fn=None): for k in range(n): sim.scan(puts_fn(k) if puts_fn else ()) # ============================================================================ def scene_new(): print() print('=' * 78) print('一、新版：功能正确性') print('=' * 78) # --- 1 单条写入 --- print('\n[1] 单条写入成功') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) run(s, 200, lambda k: [(1, 40001, 0x1234, k == 0)]) check('1 条任务成功写入', s.db['ok'] == 1 and s.db['err_cnt'] == 0, 'ok=%d err=%d' % (s.db['ok'], s.db['err_cnt'])) check('报文内容正确且只发一帧', mb.frames == [(1, 40001, 0x1234)], '%r' % (mb.frames,)) check('队列最终排空', s.db['cnt'] == 0 and s.db['empty']) # --- 2 同一周期多次调用（v2.1 核心场景）--- print('\n[2] 同一扫描周期连续调 5 次（同一个单周期脉冲喂给 5 个调用点）') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) reqs = [(1, 40001, 11), (1, 40110, 22), (2, 40050, 33), (1, 40002, 44), (3, 40099, 55)] run(s, 400, lambda k: [(a[0], a[1], a[2], True) for a in reqs] if k == 0 else []) check('5 条全部成功', s.db['ok'] == 5, 'ok=%d' % s.db['ok']) check('5 条报文一帧不落（v2.0 在这里只有第 1 条能进队列，其余静默丢失）', sorted(mb.frames) == sorted(reqs), '实发 %d 帧' % len(mb.frames)) # --- 3 同地址去重 --- print('\n[3] 同一周期把同一个寄存器连写 3 次（只有最后一次有意义）') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) run(s, 300, lambda k: [(1, 40001, 10, True), (1, 40001, 20, True), (1, 40001, 30, True)] if k == 0 else []) check('只发出 1 帧（不是 3 帧）', len(mb.frames) == 1, '实发 %d 帧' % len(mb.frames)) check('写出去的是最后一次的值 30', mb.frames == [(1, 40001, 30)], '%r' % (mb.frames,)) # --- 4 反面教材：电平直接接进来 --- print('\n[4] 反面教材：把按钮"按下位"这种电平直接接进 AddTrigger（没取沿）') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) warns = 0 for k in range(200): s.scan([(1, 40001, 7, k < 20)]) # 电平保持 20 个扫描周期 if any(o['TrigHoldWarn'] for o in s.last_outs): warns += 1 n_dup = mb.frames.count((1, 40001, 7)) check('电平保持 20 拍 -> 这个值被写了不止一次（同址去重只挡住一部分，没取沿就是错的）', n_dup >= 2, '实发 %d 帧同样的写' % n_dup) check('TrigHoldWarn 把这种接法报出来', warns >= 1, '报警 %d 拍' % warns) mb2 = MbClient('ok', resp_scans=5) o = OldSim(mb2) run(o, 200, lambda k: [(1, 40001, 7, k < 20)]) n_dup_old = mb2.frames.count((1, 40001, 7)) check('对照：原稿没有同址去重，同一条数据硬发 20 遍', n_dup_old == 20, '实发 %d 帧同样的写' % n_dup_old) mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) warns = 0 for k in range(40): trig = (k == 0) # R_TRIG 的输出：只有第一拍为 TRUE s.scan([(1, 40001, 7, trig)]) if any(o['TrigHoldWarn'] for o in s.last_outs): warns += 1 check('取沿之后按住再久也只写 1 次', len(mb.frames) == 1, '实发 %d 帧' % len(mb.frames)) check('取沿之后 TrigHoldWarn 不响', warns == 0, '报警 %d 拍' % warns) # --- 5 背压 --- print('\n[5] 队列背压：一次性塞 130 条') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3, ENABLE_WRITE=False) def flood(k): if k == 0: for i in range(130): new_put_task(s.db, 1, 40001 + i, i, True) return [] run(s, 3, flood) check('入队 99 条（容量 = 100-1）', s.db['cnt'] == 99, 'cnt=%d' % s.db['cnt']) check('拒收 31 条并记账', s.db['drop'] == 31, 'drop=%d' % s.db['drop']) check('写总闸关闭时不下发任何报文', len(mb.frames) == 0) check('WriteLocked 稳态亮起', s.db['write_locked'] is True) # --- 6 从站返回异常码：不重试 --- print('\n[6] 从站返回 Modbus 异常码 16#8383（数据地址错误/越界）') mb = MbClient('err', resp_scans=5, status_code=0x8383) s = NewSim(mb, RETRY_MAX=3) run(s, 400, lambda k: [(1, 40001, 1, k == 0)]) check('只发 1 帧就放弃（重试无意义）', len(mb.frames) == 1, '实发 %d 帧' % len(mb.frames)) check('计入 ErrCount 并锁存错误码', s.db['err_cnt'] == 1 and s.db['last_err_status'] == 0x8383, 'err=%d code=%04X' % (s.db['err_cnt'], s.db['last_err_status'])) check('失败任务快照可查（从站/地址/值）', (s.db['last_err_task'] or {}).get('addr') == 40001, '%r' % ({k: v for k, v in (s.db['last_err_task'] or {}).items()})) check('没有无意义的重试', s.db['retry'] == 0, 'retry=%d' % s.db['retry']) mb2 = MbClient('err', resp_scans=5, status_code=0x8383) o = OldSim(mb2) run(o, 400, lambda k: [(1, 40001, 1, k == 0)]) check('对照：原稿白试了 3 帧（1 次 + 2 次重试）', len(mb2.frames) == 3, '实发 %d 帧' % len(mb2.frames)) print(' （RETRY_MAX=3 沿用旧版语义：一共尝试 3 次，不是"重试 3 次"）') # --- 7 TCP 层无响应：重试 + 重连 --- print('\n[7] 从站不响应 16#80C8（TCP 层还活着）') mb = MbClient('err', resp_scans=5, status_code=0x80C8) s = NewSim(mb, RETRY_MAX=3) run(s, 1200, lambda k: [(1, 40001, 1, k == 0)]) check('重试耗尽共发 3 帧（1 次首发 + 2 次重试）', len(mb.frames) == 3, '实发 %d 帧' % len(mb.frames)) check('每次重试前断线重连', mb.disconnects >= 2, 'DISCONNECT 用了 %d 次' % mb.disconnects) check('最终丢弃并锁存 16#80C8', s.db['err_cnt'] == 1 and s.db['last_err_status'] == 0x80C8, 'err=%d code=%04X' % (s.db['err_cnt'], s.db['last_err_status'])) check('状态机回到 0，不堵死后续任务', s.db['state'] == 0, 'state=%d' % s.db['state']) # --- 8 对端彻底卡死：兜底超时救场 --- print('\n[8] 对端 TCP 半开：MB_CLIENT 受理后永远不返回') mb = MbClient('dead_busy', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) run(s, 4000, lambda k: [(1, 40001, 1, k == 0)]) check('兜底超时被触发并计数', s.db['timeout'] >= 1, 'timeout=%d' % s.db['timeout']) check('最终丢弃并锁存 16#FFFF', s.db['err_cnt'] == 1 and s.db['last_err_status'] == 0xFFFF, 'err=%d code=%04X' % (s.db['err_cnt'], s.db['last_err_status'])) check('状态机没有永久卡死', s.db['state'] == 0 and s.db['busy'] is False, 'state=%d busy=%s' % (s.db['state'], s.db['busy'])) check('超时后断开重连过', mb.disconnects >= 1, 'DISCONNECT %d 次' % mb.disconnects) mb2 = MbClient('dead_busy', resp_scans=5) s2 = NewSim(mb2, RETRY_MAX=3) def mixed(k): if k == 0: return [(1, 40001, 1, True)] if k == 4000: return [(1, 40002, 2, True)] return [] run(s2, 20000, mixed) check('卡死恢复后新任务照样走完，一条不丢', s2.db['ok'] + s2.db['err_cnt'] == 2, 'ok=%d err=%d（提交 2 条，全部结账）' % (s2.db['ok'], s2.db['err_cnt'])) # --- 9 请求未被受理 --- print('\n[9] 请求未被受理：STATUS=16#8200（同一端口有别的请求在跑）') mb = MbClient('no_accept', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) run(s, 4000, lambda k: [(1, 40001, 1, k == 0)]) check('受理超时报 16#FFFE 并计数', s.db['timeout'] >= 1 and s.db['last_err_status'] == 0xFFFE, 'timeout=%d code=%04X' % (s.db['timeout'], s.db['last_err_status'])) check('状态机没有永久卡死', s.db['state'] == 0, 'state=%d' % s.db['state']) # --- 10 参数体检 --- print('\n[10] 参数体检') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) r1 = new_put_task(s.db, 0, 40001, 1, True) # 广播号 r2 = new_put_task(s.db, 1, 70000, 1, True) # 超出地址空间 r3 = new_put_task(s.db, 1, 100, 1, True) # 不是保持寄存器区 r4 = new_put_task(s.db, 1, 40001, 1, True) # 合法 check('从站号 0 被拒，码=1', r1['Rejected'] and r1['RejectCode'] == 1) check('地址 >65535 被拒，码=2', r2['Rejected'] and r2['RejectCode'] == 2) check('地址 100 只警示不拦截', r3['AddrWarn'] and not r3['Rejected']) check('合法任务正常入队', r4['AddOk'] and not r4['Rejected']) check('队列里只有 2 条（被拒的没进去）', s.db['cnt'] == 2, 'cnt=%d' % s.db['cnt']) # --- 11 ADDR_OFFSET --- print('\n[11] ADDR_OFFSET=40001（照顾"0 基寄存器号"的写法）') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3, ADDR_OFFSET=40001) run(s, 200, lambda k: [(1, 0, 9, k == 0)]) check('0 号寄存器 -> 实际发 40001', mb.frames == [(1, 40001, 9)], '%r' % (mb.frames,)) check('该写法下不再触发地址告警', s.db['addrwarn'] == 0) # --- 12 MultiCall 诊断（v2.1：只是诊断，不是报警）--- print('\n[12] 同一周期调用 PutTask 两次：MultiCall 亮，但两条都入队') mb = MbClient('ok', resp_scans=5) s = NewSim(mb, RETRY_MAX=3) a = new_put_task(s.db, 1, 40001, 1, True) b = new_put_task(s.db, 1, 40002, 2, True) check('第 2 次调用 MultiCall 亮起（仅诊断）', a['MultiCall'] is False and b['MultiCall'] is True) check('两条都真的入队了（v2.0 这里第 2 条会丢）', a['AddOk'] is True and b['AddOk'] is True) check('队列里正好 2 条', s.db['cnt'] == 2, 'cnt=%d' % s.db['cnt']) print('\n[13] 队列满时"刚好有一格空出来"那一拍提交的写请求') ok_new = pop_scan_probe('new') ok_old = pop_scan_probe('old') check('新版这一条写进去了（判满看的是本拍现算的条目数）', ok_new is True) check('对照：原稿这一条被静默丢掉（判满用的是上一拍的标记）', ok_old is False) print(' （原稿丢的这一条没有任何计数，现场只会觉得"偶尔有一条写没生效"）') def pop_scan_probe(ver): """确定性探针：灌满队列，跑到第一次出队的那一拍，同拍提交一条写请求。 返回这条请求有没有被接收。""" mb = MbClient('ok', resp_scans=1) if ver == 'new': s = NewSim(mb, RETRY_MAX=1) db = s.db for i in range(99): new_put_task(db, 1, 40001 + i, i, True) for k in range(80): rd0 = db['rd'] new_cache_run(db, mb) popped = (db['rd'] != rd0) ok = new_put_task(db, 1, 41000 + k, k, True)['AddOk'] if popped: return ok return None o = OldSim(mb) db = o.db for i in range(99): old_put_task(db, 1, 40001 + i, i, True) for k in range(80): rd0 = db['rd'] old_cache_run(db, mb) popped = (db['rd'] != rd0) ok = old_put_task(db, 1, 41000 + k, k, True) if popped: return ok return None # ============================================================================ def scene_old(): print() print('=' * 78) print('二、对照实验：原稿的致命场景') print('=' * 78) print('\n[A] 对端 TCP 半开，MB_CLIENT 受理后永不返回') mb = MbClient('dead_busy', resp_scans=5) o = OldSim(mb) run(o, 5000, lambda k: [(1, 40001, 1, k == 0)]) check('原稿状态机永久停在 2（等完成）', o.db['state'] == 2, 'state=%d' % o.db['state']) check('原稿 Busy 永远亮着', o.db['busy'] is True) check('原稿没有任何失败计数可用于报警', o.db['sent_fail'] == 0, '失败计数=%d（一次都没记上）' % o.db['sent_fail']) run(o, 5000, lambda k: [(1, 40002 + k, k, k < 5)]) check('此后 5 条新写请求全部滞留，一帧都发不出去', len(mb.frames) == 1, '总共只发了 %d 帧' % len(mb.frames)) print('\n[B] 请求未被受理（STATUS=16#8200 端口被占用）') mb = MbClient('no_accept', resp_scans=5) o = OldSim(mb) run(o, 5000, lambda k: [(1, 40001, 1, k == 0)]) check('原稿状态机永久停在 2', o.db['state'] == 2, 'state=%d' % o.db['state']) check('原稿一帧都没发出去', len(mb.frames) == 0, '实发 %d 帧' % len(mb.frames)) check('原稿那个"重发"分支救不了它（此时 STATUS=8200 不是 7004）', o.db['resend_hits'] == 0, '该分支触发 %d 次' % o.db['resend_hits']) print('\n[C] 原稿"STATUS=7004 就重发"这个分支，到底救得了谁') mb = MbClient('dead_busy', resp_scans=5) o = OldSim(mb) run(o, 5000, lambda k: [(1, 40001, 1, k == 0)]) hits_a = o.db['resend_hits'] check('对端卡死时该分支触发 0 次', hits_a == 0, '触发 %d 次' % hits_a) mb = MbClient('no_accept', resp_scans=5) o = OldSim(mb) run(o, 5000, lambda k: [(1, 40001, 1, k == 0)]) check('请求未被受理时该分支触发 0 次', o.db['resend_hits'] == 0, '触发 %d 次' % o.db['resend_hits']) print(' （原因：16#7004 只会出现在"没有作业在跑"的时候，' '真卡死时 STATUS 是 7005 或 8200 —— 判据本身选错了）') print('\n[D] 新版对"REQ 电平会不会自动重发"两种解释都免疫') for resend in (True, False): mb = MbClient('ok', resp_scans=5, req_level_resend=resend) s = NewSim(mb, RETRY_MAX=1) run(s, 200, lambda k: [(1, 40001, 7, k == 0), (1, 40002, 8, k == 1)]) check('手册字面解释=%s 时也恰好发 2 帧（不重发）' % resend, len(mb.frames) == 2, '实发 %d 帧' % len(mb.frames)) print('\n[E] 重试口径：一条注定失败的请求要花几个来回') for label, code in (('16#8383 从站回异常码', 0x8383), ('16#818C 指针指向了优化 DB', 0x818C)): mb_o = MbClient('err', resp_scans=5, status_code=code) ob = OldSim(mb_o) run(ob, 600, lambda k: [(1, 40001, 1, k == 0)]) mb_n = MbClient('err', resp_scans=5, status_code=code) sn = NewSim(mb_n, RETRY_MAX=3) run(sn, 600, lambda k: [(1, 40001, 1, k == 0)]) check('%s：原稿 %d 帧 vs 新版 %d 帧' % (label, len(mb_o.frames), len(mb_n.frames)), len(mb_o.frames) == 3 and len(mb_n.frames) == 1) # ============================================================================ def main(): try: sys.stdout.reconfigure(encoding='utf-8') except Exception: pass scene_new() scene_old() print() print('=' * 78) print('结果：通过 %d 项，失败 %d 项' % (len(PASS), len(FAIL))) if FAIL: for f in FAIL: print(' !! %s' % f) return 1 print('全部通过。') return 0 if __name__ == '__main__': sys.exit(main())