Threaded version 0.7

component-source/Makefile

@@ -1,15 +1,16 @@
 CC ?= cc
-CFLAGS = -std=c11 -Wall -Wextra -Werror -O2
+CFLAGS = -std=c11 -Wall -Wextra -Werror -O2 -pthread
 .PHONY: test clean
 test: tests/driver tests/test_bridge tests/bridge_driver
 	./tests/test_bridge
 	python3 tests/test_tcp.py
 	python3 tests/test_tcp.py bridge_driver
-	python3 tests/test_native.py
-tests/bridge_driver: tests/bridge_driver.c src/atom_bridge.c src/atom_bridge.h src/atom_mb.c src/atom_mb.h src/atom_rtu.c src/atom_rtu.h src/atom_server.c src/atom_server.h
-	$(CC) $(CFLAGS) -Isrc tests/bridge_driver.c src/atom_bridge.c src/atom_mb.c src/atom_rtu.c src/atom_server.c -o $@
-tests/test_bridge: tests/test_bridge.c src/atom_bridge.c src/atom_bridge.h src/atom_mb.c src/atom_mb.h src/atom_rtu.c src/atom_rtu.h src/atom_server.c src/atom_server.h
-	$(CC) $(CFLAGS) -Isrc tests/test_bridge.c src/atom_bridge.c src/atom_mb.c src/atom_rtu.c src/atom_server.c -o $@
+	python3 tests/test_transport.py
+	python3 tests/test_worker.py
+tests/bridge_driver: tests/bridge_driver.c src/atom_bridge.c src/atom_bridge.h src/atom_worker.c src/atom_mb.c src/atom_mb.h src/atom_rtu.c src/atom_rtu.h src/atom_server.c src/atom_server.h
+	$(CC) $(CFLAGS) -Isrc tests/bridge_driver.c src/atom_bridge.c src/atom_worker.c src/atom_mb.c src/atom_rtu.c src/atom_server.c -o $@
+tests/test_bridge: tests/test_bridge.c src/atom_bridge.c src/atom_bridge.h src/atom_worker.c src/atom_mb.c src/atom_mb.h src/atom_rtu.c src/atom_rtu.h src/atom_server.c src/atom_server.h
+	$(CC) $(CFLAGS) -Isrc tests/test_bridge.c src/atom_bridge.c src/atom_worker.c src/atom_mb.c src/atom_rtu.c src/atom_server.c -o $@
 tests/driver: tests/driver.c src/atom_mb.c src/atom_mb.h src/atom_rtu.c src/atom_rtu.h src/atom_server.c src/atom_server.h
 	$(CC) $(CFLAGS) -Isrc tests/driver.c src/atom_mb.c src/atom_rtu.c src/atom_server.c -o $@
 clean:

component-source/component/CmpAtomModbus.c

@@ -1,3 +1,4 @@
+#define _GNU_SOURCE
 #define _DEFAULT_SOURCE
 #include "../src/atom_bridge.h"
 #include "CmpStd.h"
@@ -31,3 +32,5 @@ void CDECL CDECL_EXT atmbcall_cext(atmbcall_cext_struct *p)
 #include "../src/atom_rtu.c"
 #include "../src/atom_server.c"
 #include "../src/atom_bridge.c"
+
+#include "../src/atom_worker.c"

component-source/component/makefile

@@ -1,8 +1,8 @@
-# Generated from supplied ExtensionSDK; native build on ARM64 Pi.
+# Generated from supplied ExtensionSDK; build with target Linux toolchain.
 SDKDIR ?= ../../ExtensionSDK
 MAJORVER=0
-MINORVER=3
-CMPVERSION=0x00030000
+MINORVER=7
+CMPVERSION=0x00040000
 # Local development ID; check no other component uses this ID.
 CMPID=0x2001
 DOS2UNIX=dos2unix
@@ -10,4 +10,6 @@ M4=m4
 CC ?= gcc
 # The SDK otherwise includes CmpFrame in its component identity wildcard.
 override CSRC := CmpAtomModbus.c
+CFLAGS += -pthread
+LDLIBS += -pthread
 include $(SDKDIR)/makefile

component-source/src/atom_bridge.c

@@ -4,7 +4,7 @@
 #include "atom_server.h"
 #include <stdio.h>
 #include <string.h>
-typedef struct { am_client c; ar_transport rtu; int serial; uint32_t generation; int used, retired; } slot;
+typedef struct { am_client c; ar_transport rtu; int serial; uint32_t generation; int used, retired, suspended; uint64_t heartbeat; } slot;
 static slot slots[AM_POOL_SIZE];
 static uint32_t token(unsigned i) { return (slots[i].generation<<8)|(i+1u); }
 static slot *lookup(uint32_t h) {
@@ -24,7 +24,7 @@ static void snapshot(const am_client *c,uint32_t h,uint32_t *s) {
  s[18]=c->function; s[19]=c->unit; s[20]=(uint32_t)c->enabled;
  s[21]=c->connect_timeout_ms; s[22]=c->response_timeout_ms; s[23]=c->retry_delay_ms;
 }
-uint32_t ab_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
+uint32_t ab_dispatch(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
  uint8_t unit,uint8_t f,uint16_t addr,uint16_t q,uint16_t *data,uint32_t cap,
  uint32_t *status,uint32_t sc,uint64_t now,uint32_t ct,uint32_t rt,uint32_t retry) {
  slot *s; unsigned i; uint32_t result=0;
@@ -47,7 +47,7 @@ uint32_t ab_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
    text[q]=0;
    if(!ar_init(&s->c,&s->rtu,text,ip,unit,f,(uint8_t)addr,ct))return AM_ARGUMENT;
   } else if(!am_init(&s->c,text,port)) return AM_ARGUMENT;
-  s->generation++; s->used=1; h=token(i);
+  s->generation++; s->used=1; s->suspended=0; s->heartbeat=now; h=token(i);
   snapshot(&s->c,h,status); return 0;
  }
  s=lookup(h);
@@ -58,9 +58,12 @@ uint32_t ab_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
   /* Never recycle a generation after wrap: stale handles stay invalid. */
   if(s->generation==0xffffffu) s->retired=1;
   snapshot(&s->c,0,status); return 0;
- case AB_ENABLE: am_enable(&s->c,1,now); break;
- case AB_DISABLE: if(s->serial)ar_close(&s->c);else am_enable(&s->c,0,now); break;
- case AB_TICK: if(s->serial)ar_tick(&s->c,&s->rtu,now);else am_tick(&s->c,now); break;
+ case AB_ENABLE: s->heartbeat=now; s->suspended=0; am_enable(&s->c,1,now); break;
+ case AB_DISABLE: s->suspended=0; if(s->serial)ar_close(&s->c);else am_enable(&s->c,0,now); break;
+ case AB_TICK:
+  s->heartbeat=now;
+  if(s->suspended) { s->suspended=0; am_enable(&s->c,1,now); }
+  break;
  case AB_RECONNECT: if(s->serial)ar_reconnect(&s->c,&s->rtu,now);else am_reconnect(&s->c,now); break;
  case AB_CONFIGURE:
   if(s->c.enabled) result=AM_BUSY;
@@ -68,6 +71,7 @@ uint32_t ab_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
   else { s->c.connect_timeout_ms=ct; s->c.response_timeout_ms=rt; s->c.retry_delay_ms=retry; }
   break;
  case AB_SUBMIT:
+  if(s->c.done || s->c.error) { result=AM_BUSY; break; }
   if((f==5 || f==6 || f==15 || f==16) && (!data || cap<q)) result=AB_BAD_CAPACITY;
   else result=(uint32_t)(s->serial?ar_request(&s->c,&s->rtu,unit,f,addr,q,data,now):am_request(&s->c,unit,f,addr,q,data,now));
   break;
@@ -76,6 +80,7 @@ uint32_t ab_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
   else if(!data || cap<s->c.quantity) result=AB_BAD_CAPACITY;
   else memcpy(data,s->c.data,s->c.quantity*sizeof(*data));
   break;
+ case 12: break;
  case AB_ACK: s->c.done=0; s->c.error=0; break;
  case AB_REQUEST_TIMEOUT:
   if(s->c.busy) result=AM_BUSY;
@@ -84,9 +89,15 @@ uint32_t ab_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,
   break;
  default: result=AB_BAD_COMMAND; break;
  }
- snapshot(&s->c,h,status); return result;
+ snapshot(&s->c,h,status);
+ if(cmd==12){
+  status[21]=(uint32_t)(s->c.completed_ms>=s->c.started_ms?s->c.completed_ms-s->c.started_ms:0);
+  uint64_t age=s->c.has_success?now-s->c.last_success_ms:0;
+  status[22]=age>UINT32_MAX?UINT32_MAX:(uint32_t)age;
+ }
+ return result;
 }
-void ab_shutdown(void) {
+void ab_close_all(void) {
  as_shutdown();
  unsigned i;
  for(i=0;i<AM_POOL_SIZE;i++) if(slots[i].used) {
@@ -94,3 +105,16 @@ void ab_shutdown(void) {
   if(slots[i].generation==0xffffffu) slots[i].retired=1;
  }
 }
+
+/* Called only by the worker while holding the shared engine mutex. */
+void ab_pump(uint64_t now,uint32_t watchdog_ms) {
+ for(unsigned i=0;i<AM_POOL_SIZE;i++) if(slots[i].used) {
+  slot *s=&slots[i];
+  if(s->c.enabled && now-s->heartbeat >= watchdog_ms) {
+   if(s->serial)ar_close(&s->c);else am_close(&s->c);
+   s->suspended=1;
+  }
+  if(s->serial)ar_tick(&s->c,&s->rtu,now);else am_tick(&s->c,now);
+ }
+ as_pump(now,watchdog_ms);
+}

component-source/src/atom_bridge.h

@@ -1,14 +1,14 @@
 #ifndef ATOM_BRIDGE_H
 #define ATOM_BRIDGE_H
 #include <stdint.h>
-/* Fixed pool, one IEC task owns ALL calls. Do not share across tasks.
+/* Fixed pool; serialized calls and worker I/O. IEC instances still belong to one task.
  * Only synchronous pointer access; generation-checked handles, no heap. */
 #define AM_STATUS_WORDS 24
 #define AM_POOL_SIZE 16
 enum { AB_OPEN=1, AB_CLOSE, AB_ENABLE, AB_DISABLE, AB_TICK, AB_SUBMIT,
  AB_COPY_READ, AB_RECONNECT, AB_CONFIGURE, AB_ACK, AB_REQUEST_TIMEOUT };
 enum { AB_OPEN_RTU=20 };
-enum { AB_BAD_HANDLE=256, AB_POOL_FULL, AB_BAD_CAPACITY, AB_BAD_COMMAND };
+enum { AB_BAD_HANDLE=256, AB_POOL_FULL, AB_BAD_CAPACITY, AB_BAD_COMMAND, AB_WORKER_ERROR };
 uint32_t ab_call(uint32_t command,uint32_t handle,uint32_t ip,uint16_t port,
  uint8_t unit,uint8_t function,uint16_t address,uint16_t quantity,
  uint16_t *data,uint32_t capacity,uint32_t *status,uint32_t status_capacity,

component-source/src/atom_mb.c

@@ -5,6 +5,7 @@
 #include <poll.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <netinet/tcp.h>
 #include <sys/ioctl.h>
 #include <unistd.h>
 #ifndef MSG_NOSIGNAL
@@ -14,12 +15,12 @@ static uint16_t get16(const uint8_t *p) { return (uint16_t)((p[0]<<8)|p[1]); }
 static void put16(uint8_t *p,uint16_t v) { p[0]=(uint8_t)(v>>8); p[1]=(uint8_t)v; }
 static int writing(uint8_t f) { return f==5 || f==6 || f==15 || f==16; }
 static void drop(am_client *c) { if(c->fd>=0) { if(c->serial_port)ioctl(c->fd,TIOCNXCL);close(c->fd); } c->fd=-1; }
-static void finish(am_client *c,am_error e) {
- c->busy=0; c->done=e==AM_OK; c->error=e!=AM_OK; c->result=e;
+static void finish(am_client *c,am_error e,uint64_t now) {
+ c->completed_ms=now; c->busy=0; c->done=e==AM_OK; c->error=e!=AM_OK; c->result=e;
  if(e==AM_OK) c->completed++; else { c->failed++; c->last_error=e; }
 }
 static void fault(am_client *c,am_error e,uint64_t now) {
- if(c->busy) { c->write_uncertain=writing(c->function) && c->tx_pos>0; finish(c,e); }
+ if(c->busy) { c->write_uncertain=writing(c->function) && c->tx_pos>0; finish(c,e,now); }
  c->last_error=e; drop(c); c->state=c->enabled?AM_RETRY:AM_DISABLED;
  c->deadline_ms=now+c->retry_delay_ms;
 }
@@ -63,7 +64,7 @@ am_error am_request(am_client *c,uint8_t unit,uint8_t f,uint16_t addr,uint16_t q
    else if(v[i]) c->tx[13+i/8]|=(uint8_t)(1u<<(i%8));
   }
  }
- c->unit=unit; c->function=f; c->address=addr; c->quantity=q;
+ c->started_ms=now; c->unit=unit; c->function=f; c->address=addr; c->quantity=q;
  c->tx_len=n; c->tx_pos=0; c->rx_len=0; c->rx_need=7;
  c->busy=1; c->done=0; c->error=0; c->exception=0; c->write_uncertain=0;
  c->result=AM_OK; c->state=AM_SENDING; c->deadline_ms=now+c->response_timeout_ms;
@@ -73,7 +74,7 @@ void am_accept_response(am_client *c,uint64_t now) {
  unsigned i; size_t bytes;
  if(c->rx[7]==(uint8_t)(c->function|0x80)) {
   if(c->rx_need!=9 || !c->rx[8]) { fault(c,AM_PROTOCOL,now); return; }
-  c->exception=c->rx[8]; finish(c,AM_EXCEPTION); c->state=AM_READY; return;
+  c->exception=c->rx[8]; finish(c,AM_EXCEPTION,now); c->state=AM_READY; return;
  }
  if(c->rx[7]!=c->function) { fault(c,AM_PROTOCOL,now); return; }
  if(writing(c->function)) {
@@ -84,7 +85,7 @@ void am_accept_response(am_client *c,uint64_t now) {
   for(i=0;i<c->quantity;i++) c->data[i]=c->function<=2?
    (uint16_t)((c->rx[9+i/8]>>(i%8))&1):get16(c->rx+9+2*i);
  }
- c->last_success_ms=now; c->has_success=1; finish(c,AM_OK); c->state=AM_READY;
+ c->last_success_ms=now; c->has_success=1; finish(c,AM_OK,now); c->state=AM_READY;
 }
 void am_tick(am_client *c,uint64_t now) {
  struct sockaddr_in a; struct pollfd p; int e,r; socklen_t len; ssize_t n;
@@ -92,8 +93,10 @@ void am_tick(am_client *c,uint64_t now) {
  if(c->state==AM_RETRY) {
   if(now<c->deadline_ms) return;
   c->connect_attempts++; if(c->connect_attempts>1) c->reconnects++;
-  c->socket_error=0; c->fd=socket(AF_INET,SOCK_STREAM,0);
+  c->socket_error=0; c->fd=socket(AF_INET,SOCK_STREAM|SOCK_CLOEXEC,0);
   if(c->fd<0) { c->socket_error=errno; fault(c,AM_SOCKET,now); return; }
+  int one=1;
+  setsockopt(c->fd,IPPROTO_TCP,TCP_NODELAY,&one,sizeof(one));
   r=fcntl(c->fd,F_GETFL,0);
   if(r<0 || fcntl(c->fd,F_SETFL,r|O_NONBLOCK)<0) { c->socket_error=errno; fault(c,AM_SOCKET,now); return; }
   memset(&a,0,sizeof(a)); a.sin_family=AF_INET; a.sin_port=htons(c->port);

component-source/src/atom_mb.h

@@ -15,7 +15,7 @@ typedef struct {
  uint8_t exception, unit, function;
  uint16_t transaction, address, quantity;
  uint32_t connect_timeout_ms, response_timeout_ms, retry_delay_ms;
- uint64_t deadline_ms, last_success_ms;
+ uint64_t deadline_ms, last_success_ms, started_ms, completed_ms;
  uint32_t connect_attempts, reconnects, completed, failed;
  int busy, done, error, write_uncertain, has_success;
  char ip[16]; uint16_t port;

component-source/src/atom_rtu.c

@@ -29,7 +29,7 @@ int ar_init(am_client *c,ar_transport *r,const char *path,uint32_t baud,uint8_t
  return 1;
 }
 static void ar_fault(am_client *c,ar_transport *r,am_error error,uint64_t now) {
- if(c->busy) { c->write_uncertain=(c->function>=5 && r->tx_pos>0);c->busy=0;c->done=0;c->error=1;c->result=error;c->failed++; }
+ if(c->busy) { c->write_uncertain=(c->function>=5 && r->tx_pos>0);c->completed_ms=now;c->busy=0;c->done=0;c->error=1;c->result=error;c->failed++; }
  c->last_error=error;if(c->fd>=0){ioctl(c->fd,TIOCNXCL);close(c->fd);}c->fd=-1;c->state=c->enabled?AM_RETRY:AM_DISABLED;
  c->deadline_ms=now+c->retry_delay_ms;r->quiet_until=c->deadline_ms;r->rx_len=0;
 }

component-source/src/atom_server.c

@@ -14,8 +14,8 @@
 #endif
 typedef struct { int fd;uint8_t rx[260],tx[260];size_t received,needed,sent,length;uint64_t deadline; } as_peer;
 typedef struct {
- int used,enabled,listener,running,error,os_error;uint32_t generation,received,replied,exceptions,retry;
- uint32_t ip;uint16_t port,base[4],count[4],values[4][AS_ITEMS];uint64_t retry_at;
+ int used,enabled,suspended,listener,running,error,os_error;uint32_t generation,received,replied,exceptions,retry;
+ uint32_t ip;uint16_t port,base[4],count[4],values[4][AS_ITEMS];uint64_t retry_at,heartbeat; uint8_t dirty[4][AS_ITEMS];
  as_peer peers[AS_PEERS];
 } as_server;
 static as_server as_slots[AS_POOL];
@@ -46,13 +46,13 @@ static void as_reply(as_server *s,as_peer *p){
   table=f==5?0:2;
   if(p->needed!=12 || (f==5 && q!=0 && q!=0xff00)){exception=3;break;}
   if(!as_range(s,table,address,1)){exception=2;break;}
-  s->values[table][address-s->base[table]]=f==5?(q!=0):q;memcpy(p->tx+8,p->rx+8,4);n=12;break;
+  s->values[table][address-s->base[table]]=f==5?(q!=0):q;s->dirty[table][address-s->base[table]]=1;memcpy(p->tx+8,p->rx+8,4);n=12;break;
  case 15:case 16:
   table=f==15?0:2;
   if(!q || q>(f==15?1968:123) || p->needed<13 || p->rx[12]!=(f==15?(q+7u)/8u:2u*q) || p->needed!=(size_t)(13+p->rx[12])){exception=3;break;}
   if(!as_range(s,table,address,q)){exception=2;break;}
   index=address-s->base[table];
-  for(unsigned i=0;i<q;i++)s->values[table][index+i]=f==15?(uint16_t)((p->rx[13+i/8]>>(i%8))&1):as_get16(p->rx+13+2*i);
+  for(unsigned i=0;i<q;i++){s->values[table][index+i]=f==15?(uint16_t)((p->rx[13+i/8]>>(i%8))&1):as_get16(p->rx+13+2*i);s->dirty[table][index+i]=1;}
   memcpy(p->tx+8,p->rx+8,4);n=12;break;
  default:exception=1;break;
  }
@@ -103,7 +103,7 @@ uint32_t as_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,uint8_t table
   memset(out,0,24*sizeof(*out));if(!port || !retry)return AM_ARGUMENT;
   for(i=0;i<AS_POOL;i++)if(!as_slots[i].used && as_slots[i].generation<0xffffffu)break;
   if(i==AS_POOL)return 257;
-  s=&as_slots[i];uint32_t generation=s->generation+1;memset(s,0,sizeof(*s));s->generation=generation;s->used=1;s->listener=-1;s->ip=ip;s->port=port;s->retry=retry;
+  s=&as_slots[i];uint32_t generation=s->generation+1;memset(s,0,sizeof(*s));s->generation=generation;s->used=1;s->listener=-1;s->ip=ip;s->port=port;s->retry=retry;s->heartbeat=now;
   for(unsigned j=0;j<AS_PEERS;j++){s->peers[j].fd=-1;as_peer_close(&s->peers[j]);}
   h=(generation<<8)|(i+65u);as_snapshot(s,h,out);return 0;
  }
@@ -112,20 +112,32 @@ uint32_t as_call(uint32_t cmd,uint32_t h,uint32_t ip,uint16_t port,uint8_t table
  s=&as_slots[i-1];
  switch(cmd){
  case AS_CLOSE:as_stop(s);s->used=0;as_snapshot(s,0,out);return 0;
- case AS_ENABLE:s->enabled=1;break;
- case AS_DISABLE:s->enabled=0;as_stop(s);break;
- case AS_TICK:as_tick(s,now);break;
+ case AS_ENABLE:s->enabled=1;s->suspended=0;s->heartbeat=now;break;
+ case AS_DISABLE:s->suspended=0;s->enabled=0;as_stop(s);break;
+ case AS_TICK:s->heartbeat=now;if(s->suspended){s->suspended=0;s->enabled=1;s->retry_at=now;}break;
  case AS_PUSH:case AS_PULL:
   if(table>3 || !q || q>AS_ITEMS || (uint32_t)addr+q>65536u)result=AM_ARGUMENT;
   else if(!data || cap<q)result=258;
   else if(cmd==AS_PUSH){
    if(table<2){for(unsigned j=0;j<q;j++)if(data[j]>1){result=AM_ARGUMENT;break;}}
-   if(!result){s->base[table]=addr;s->count[table]=q;memcpy(s->values[table],data,q*sizeof(*data));}
+   if(!result){
+    if(s->count[table] && (s->base[table]!=addr || s->count[table]!=q)) { result=AM_ARGUMENT;break; }
+    s->base[table]=addr;s->count[table]=q;
+    for(unsigned j=0;j<q;j++)if(!s->dirty[table][j])s->values[table][j]=data[j];
+   }
   }else if(s->base[table]!=addr || s->count[table]!=q)result=AM_ARGUMENT;
-  else memcpy(data,s->values[table],q*sizeof(*data));
+  else {memcpy(data,s->values[table],q*sizeof(*data));memset(s->dirty[table],0,q);}
   break;
  default:result=259;break;
  }
  as_snapshot(s,h,out);return result;
 }
 void as_shutdown(void){for(unsigned i=0;i<AS_POOL;i++)if(as_slots[i].used){as_stop(&as_slots[i]);as_slots[i].used=0;}}
+
+void as_pump(uint64_t now,uint32_t watchdog_ms){
+ for(unsigned i=0;i<AS_POOL;i++)if(as_slots[i].used){
+  as_server *s=&as_slots[i];
+  if(s->enabled && now-s->heartbeat>=watchdog_ms){as_stop(s);s->enabled=0;s->suspended=1;}
+  as_tick(s,now);
+ }
+}

component-source/src/atom_server.h

@@ -5,3 +5,5 @@ enum { AS_OPEN=30,AS_CLOSE,AS_ENABLE,AS_DISABLE,AS_TICK,AS_PUSH,AS_PULL };
 uint32_t as_call(uint32_t,uint32_t,uint32_t,uint16_t,uint8_t,uint16_t,uint16_t,uint16_t *,uint32_t,uint32_t *,uint64_t,uint32_t);
 void as_shutdown(void);
 #endif
+
+void as_pump(uint64_t now,uint32_t watchdog_ms);

component-source/tests/test_native.py

@@ -1,114 +0,0 @@
-#!/usr/bin/env python3
-"""Real PTY/socket integration tests for the native RTU and TCP server bridge."""
-import ctypes as C,os,pty,select,socket,struct,subprocess,time,tempfile
-from pathlib import Path
-root=Path(__file__).resolve().parents[1]
-with tempfile.TemporaryDirectory() as tmp:
- so=Path(tmp)/'test-native.so'
- subprocess.run(['cc','-shared','-fPIC','-std=c11','-Wall','-Wextra','-Werror','-O2','-I'+str(root/'src'),*[str(root/'src'/f) for f in ['atom_mb.c','atom_rtu.c','atom_server.c','atom_bridge.c']],'-o',str(so)],check=True)
- lib=C.CDLL(str(so));fn=lib.ab_call
- U=C.c_uint32;W=C.c_uint16;B=C.c_uint8
- fn.argtypes=[U,U,U,W,B,B,W,W,C.POINTER(W),U,C.POINTER(U),U,C.c_uint64,U,U,U];fn.restype=U
- data=(W*2000)();status=(U*24)()
- def now():return int(time.monotonic()*1000)
- def call(cmd,h=0,ip=0,port=0,unit=1,f=3,addr=0,q=1,ct=100,rt=100,retry=10,when=None):
-  return fn(cmd,h,ip,port,unit,f,addr,q,data,2000,status,24,now() if when is None else when,ct,rt,retry)
- def crc(frame):
-  v=0xffff
-  for byte in frame:
-   v^=byte
-   for _ in range(8):v=(v>>1)^(0xa001 if v&1 else 0)
-  return v
- def adu(payload):return payload+struct.pack('<H',crc(payload))
- def ready(h):
-  end=time.monotonic()+1
-  while time.monotonic()<end:
-   assert call(5,h)==0
-   if status[1]==3:return
-   time.sleep(.001)
-  raise AssertionError(('not ready',list(status)))
- def rtu_scenario(mode,f=3):
-  master,slave=pty.openpty();path=os.ttyname(slave);os.close(slave);os.set_blocking(master,False)
-  for i,b in enumerate(path.encode()):data[i]=b
-  assert call(20,ip=9600,unit=0,f=8,addr=1,q=len(path),ct=0)==0
-  h=status[0];assert call(9,h)==0;assert call(3,h)==0;ready(h)
-  for attempt in range(2 if mode in ['crc','timeout','echo'] else 1):
-   if f==16:data[0]=123;data[1]=456
-   if f==15:data[0]=1;data[1]=0
-   assert call(6,h,unit=1,f=f,addr=10,q=2)==0
-   received=b'';sent=False;fragments=[];deadline=time.monotonic()+1
-   while time.monotonic()<deadline:
-    assert call(5,h)==0
-    if select.select([master],[],[],0)[0]:
-     try:received+=os.read(master,256)
-     except OSError:pass
-    need=(9+received[6]) if len(received)>=7 and f>=15 else 8
-    if len(received)>=need and not sent:
-     assert crc(received)==0;sent=True
-     payload=(bytes([1,f])+received[2:6]) if f>=15 else bytes([1,f,4,0,42,0,43])
-     if mode=='exception':payload=bytes([1,f|0x80,2])
-     if mode=='echo' and attempt==0:payload=payload[:3]+bytes([payload[3]^1])+payload[4:]
-     response=adu(payload)
-     if mode=='crc' and attempt==0:response=response[:-1]+bytes([response[-1]^1])
-     if mode!='timeout' or attempt>0:
-      fragments=[response[:2],response[2:4],response[4:]] if mode=='fragment' else [response]
-    if fragments:os.write(master,fragments.pop(0))
-    if status[3] or status[4]:break
-    time.sleep(.002)
-   assert sent and (status[3] or status[4]),list(status)
-   if mode in ['crc','echo'] and attempt==0:assert status[5]==8
-   elif mode=='timeout' and attempt==0:assert status[5]==6
-   elif mode=='exception':assert status[5]==9 and status[8]==2 and status[1]==3
-   else:
-    assert status[3]==1,list(status)
-    if f<=4:assert call(7,h)==0 and list(data[:2])==[42,43]
-   assert call(10,h)==0
-   if attempt==0 and mode in ['crc','timeout','echo']:ready(h)
-  assert call(6,h,unit=0,f=3,addr=0,q=1)==1 # Broadcast unsupported, no fake completion.
-  assert call(2,h)==0;assert call(5,h)==256;os.close(master)
- for mode in ['normal','fragment','crc','timeout','exception']:rtu_scenario(mode);print('PASS RTU',mode)
- rtu_scenario('normal',16);print('PASS RTU register write')
- rtu_scenario('echo',16);print('PASS RTU invalid echo and reopen recovery')
- rtu_scenario('normal',15);print('PASS RTU coil write')
- # Native TCP server, four independent tables mapped at nonzero Modbus address.
- probe=socket.socket();probe.bind(('127.0.0.1',0));port=probe.getsockname()[1];probe.close()
- assert call(30,ip=0x7f000001,port=port)==0;h=status[0]
- for table in range(4):
-  for i in range(16):data[i]=i%2 if table<2 else 100+i
-  assert call(35,h,f=table,addr=10,q=16)==0
- assert call(32,h)==0;assert call(34,h)==0 and status[1]==1
- peer=socket.create_connection(('127.0.0.1',port));peer.setblocking(False)
- tid=0
- def exchange(pdu):
-  global tid
-  tid+=1;packet=struct.pack('>HHHB',tid,0,len(pdu)+1,1)+pdu
-  peer.send(packet[:3]);assert call(34,h)==0;peer.send(packet[3:])
-  reply=b'';end=time.monotonic()+1
-  while time.monotonic()<end:
-   assert call(34,h)==0
-   try:reply+=peer.recv(260)
-   except BlockingIOError:pass
-   if len(reply)>=7 and len(reply)==6+struct.unpack('>H',reply[4:6])[0]:return reply[7:]
-   time.sleep(.001)
-  raise AssertionError(('server timeout',pdu,reply))
- for f in [1,2,3,4]:
-  rep=exchange(struct.pack('>BHH',f,10,2));assert rep==(bytes([f,1,2]) if f<=2 else bytes([f,4,0,100,0,101]));print('PASS slave read FC',f)
- for f in [5,6]:
-  value=0xff00 if f==5 else 456;pdu=struct.pack('>BHH',f,10,value);assert exchange(pdu)==pdu
-  assert call(36,h,f=0 if f==5 else 2,addr=10,q=16)==0 and data[0]==(1 if f==5 else 456);print('PASS slave write FC',f)
- for f in [15,16]:
-  payload=b'\x01' if f==15 else struct.pack('>HH',321,654)
-  pdu=struct.pack('>BHHB',f,10,2,len(payload))+payload
-  assert exchange(pdu)==struct.pack('>BHH',f,10,2)
-  assert call(36,h,f=0 if f==15 else 2,addr=10,q=16)==0 and list(data[:2])==([1,0] if f==15 else [321,654]);print('PASS slave write FC',f)
- assert exchange(struct.pack('>BHH',3,0,1))==bytes([0x83,2])
- assert exchange(struct.pack('>BHH',16,10,0))==bytes([0x90,3])
- assert exchange(struct.pack('>BHH',99,10,1))==bytes([0xe3,1])
- print('PASS slave exceptions and nonzero table bases')
- assert call(31,h)==0;assert call(34,h)==256;peer.close()
- # Listener bind error recovers after the conflicting listener disappears.
- blocker=socket.socket();blocker.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1);blocker.bind(('127.0.0.1',port));blocker.listen()
- assert call(30,ip=0x7f000001,port=port)==0;h=status[0];call(32,h);call(34,h);assert status[2]==1
- blocker.close();call(34,h,when=now()+100);assert status[1]==1 and status[2]==0
- call(31,h);print('PASS slave listener recovery and stale handles')
- lib.ab_shutdown()