Preserve separate Modbus phase timeouts

Connect and write operations each have a one-second limit, and the
response gets two seconds after the write completes. Caller deadlines
remain earlier bounds. Deadline changes are serialized with context
interruption so a read cannot replace the cancellation deadline. A
simulated delayed write verifies the complete response window.

Issue: 8fdc39a

diff --git a/doc/m31.rst b/doc/m31.rst
index 64c913c6312b4e4fb3b5f8bfe690555071f14bc8..b0305e346d0131f7b048553666a695c89d404b16 100644 (file)
--- a/doc/m31.rst
+++ b/doc/m31.rst
@@ -33,10 +33,10 @@ load. The conservative sixteen-module limit includes the host.
 ***************
 
 ``NewTCPClient`` supplies a caller-owned, serialized connection through
 ***************
 
 ``NewTCPClient`` supplies a caller-owned, serialized connection through

-``Client``. Connect selects the configured unit and has a two-second
-timeout. A request has a one-second write deadline and a two-second
-response deadline, both measured from its start. Caller deadlines and
-cancellation can stop it earlier. Each response must match the
+``Client``. Connect selects the configured unit and has a one-second
+timeout. Each write has a one-second deadline. Each response has a
+two-second deadline measured after the request write. Caller deadlines
+and cancellation can stop it earlier. Each response must match the

 transaction, unit, function, length, and write acknowledgement.
 Supported functions are 01, 02, 03, 05, 15, and 16. No external Modbus
 package is required.
 transaction, unit, function, length, and write acknowledgement.
 Supported functions are 01, 02, 03, 05, 15, and 16. No external Modbus
 package is required.

diff --git a/driver/m31/tcp_client.go b/driver/m31/tcp_client.go
index 972f6984e519180badad7959ada2e1ae16d0bebd..7ee6bbe3368bc6865c301d3ab1c17a7d81c209a5 100644 (file)
--- a/driver/m31/tcp_client.go
+++ b/driver/m31/tcp_client.go
@@ -9,11 +9,12 @@ import (
        "io"
        "net"
        "os"
        "io"
        "net"
        "os"

+       "sync"

        "time"
 )
 
 const (
        "time"
 )
 
 const (

-       connectTimeout = 2 * time.Second
+       connectTimeout = time.Second

        writeTimeout   = time.Second
        readTimeout    = 2 * time.Second
 )
        writeTimeout   = time.Second
        readTimeout    = 2 * time.Second
 )


@@ -25,7 +26,7 @@ type tcpClient struct {
 }
 
 // NewTCPClient supplies Modbus TCP functions 01, 02, 03, 05, 15, and 16 without
 }
 
 // NewTCPClient supplies Modbus TCP functions 01, 02, 03, 05, 15, and 16 without

-// opening a connection. Connect has a two-second timeout. Writes have a
+// opening a connection. Connect has a one-second timeout. Writes have a

 // one-second timeout, and reads have a two-second timeout. Earlier caller
 // deadlines apply to each operation. The caller owns Close.
 func NewTCPClient() Client {
 // one-second timeout, and reads have a two-second timeout. Earlier caller
 // deadlines apply to each operation. The caller owns Close.
 func NewTCPClient() Client {


@@ -187,16 +188,17 @@ func (c *tcpClient) exchange(ctx context.Context, function byte, data []byte) (r
                return nil, fmt.Errorf("set Modbus write deadline: %w", err)
        }
 
                return nil, fmt.Errorf("set Modbus write deadline: %w", err)
        }
 

-       err = c.conn.SetReadDeadline(operationDeadline(ctx, readTimeout))
-       if err != nil {
-               return nil, fmt.Errorf("set Modbus read deadline: %w", err)
-       }
-

        interrupted := make(chan struct{})
 
        interrupted := make(chan struct{})
 

-       var interruptErr error
+       var (
+               interruptErr error
+               deadlineMu   sync.Mutex
+       )

 
        stop := context.AfterFunc(ctx, func() {
 
        stop := context.AfterFunc(ctx, func() {

+               deadlineMu.Lock()
+               defer deadlineMu.Unlock()
+

                interruptErr = c.conn.SetDeadline(time.Now())
 
                close(interrupted)
                interruptErr = c.conn.SetDeadline(time.Now())
 
                close(interrupted)


@@ -222,6 +224,21 @@ func (c *tcpClient) exchange(ctx context.Context, function byte, data []byte) (r
                return nil, fmt.Errorf("write Modbus request: %w", io.ErrShortWrite)
        }
 
                return nil, fmt.Errorf("write Modbus request: %w", io.ErrShortWrite)
        }
 

+       deadlineMu.Lock()
+
+       if ctx.Err() != nil {
+               deadlineMu.Unlock()
+
+               return nil, ctx.Err()
+       }
+
+       err = c.conn.SetReadDeadline(operationDeadline(ctx, readTimeout))
+       deadlineMu.Unlock()
+
+       if err != nil {
+               return nil, fmt.Errorf("set Modbus read deadline: %w", err)
+       }
+

        var header [7]byte
 
        _, err = io.ReadFull(c.conn, header[:])
        var header [7]byte
 
        _, err = io.ReadFull(c.conn, header[:])

diff --git a/driver/m31/tcp_client_test.go b/driver/m31/tcp_client_test.go
index 6e4979b80ac81d443505bda5f0ca41f5bb2c2e1a..0067566b070cf21907b8b465479db9123d1e432a 100644 (file)
--- a/driver/m31/tcp_client_test.go
+++ b/driver/m31/tcp_client_test.go
@@ -311,6 +311,59 @@ func TestTCPClientUsesCallerDeadlineAndStartsClosed(t *testing.T) {
        })
 }
 
        })
 }
 

+func TestTCPClientReadTimeoutStartsAfterRequestWrite(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               conn, peer := net.Pipe()
+               c := &tcpClient{conn: conn, unit: 1}
+               t.Cleanup(func() {
+                       require.NoError(t, c.Close())
+                       require.NoError(t, peer.Close())
+               })
+
+               done := make(chan error, 1)
+               go func() {
+                       time.Sleep(500 * time.Millisecond)
+
+                       var request [12]byte
+
+                       _, err := io.ReadFull(peer, request[:])
+                       done <- err
+               }()
+
+               started := time.Now()
+               _, err := c.ReadCoils(t.Context(), 1)
+               require.Error(t, err)
+               require.NoError(t, <-done)
+               require.Equal(t, 2500*time.Millisecond, time.Since(started))
+       })
+}
+
+func TestTCPClientStopsAfterCancelledRequestWrite(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               conn, peer := net.Pipe()
+               t.Cleanup(func() {
+                       require.NoError(t, peer.Close())
+               })
+
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               transport := &interruptConn{
+                       Conn:        conn,
+                       cancel:      cancel,
+                       interrupted: make(chan struct{}),
+               }
+               c := &tcpClient{conn: transport, unit: 1}
+               t.Cleanup(func() {
+                       require.NoError(t, c.Close())
+               })
+
+               _, err := c.ReadCoils(ctx, 1)
+               require.ErrorIs(t, err, context.Canceled)
+               require.NotErrorIs(t, err, os.ErrDeadlineExceeded)
+       })
+}
+

 func TestTCPClientCancellationPreservesIndependentFailure(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
                conn, peer := net.Pipe()
 func TestTCPClientCancellationPreservesIndependentFailure(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
                conn, peer := net.Pipe()


@@ -363,11 +416,15 @@ type interruptConn struct {
        interrupted chan struct{}
 }
 
        interrupted chan struct{}
 }
 

-func (c *interruptConn) Write([]byte) (int, error) {
+func (c *interruptConn) Write(data []byte) (int, error) {

        c.cancel()
        <-c.interrupted
 
        c.cancel()
        <-c.interrupted
 

-       return 0, c.failure
+       if c.failure != nil {
+               return 0, c.failure
+       }
+
+       return len(data), nil

 }
 
 func (c *interruptConn) SetDeadline(deadline time.Time) error {
 }
 
 func (c *interruptConn) SetDeadline(deadline time.Time) error {