This is an automated email from the ASF dual-hosted git repository. sruehl pushed a commit to branch develop in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 04527e01c3aa1845d6a47d2d84527f819cf47e5c Author: Sebastian Rühl <[email protected]> AuthorDate: Wed Jul 15 09:16:44 2026 +0200 fix(plc4go): never block result delivery to abandoned request channels Drivers hand results to the caller over a channel with capacity 1 whose consumer reads at most one result - and a caller whose context expired reads none. Any surplus send therefore blocks its goroutine forever once the buffer slot is taken. Field goroutine dumps (gateway stuck for days) show exactly this chain on modbus: 1. the poll caller times out and abandons the result channel 2. SendRequest fails against the dead device; the send-error result fills the single buffer slot 3. the expectation registered by SendRequest is left behind and fires its error handler on timeout 60s later - that second send blocks forever, pinned in the codec's WaitGroup (378 accumulated) 4. connection close -> defaultCodec.Disconnect -> wg.Wait() hangs indefinitely, wedging the caller's shutdown/restart path Fixes, each sufficient on its own for the observed chain: - spi/default: SendRequest removes its expectation again when Send fails; the caller already got the error directly, a later timeout would double-fire the error handler for the same request. - spi/utils: new DeliverResult helper that drops a result (with a warning) instead of blocking when the channel is full and abandoned. - modbus, bacnetip: deliver all read/write/subscribe results through DeliverResult. The modbus Writer additionally delivers the send error as a result instead of only logging it - callers without their own deadline otherwise wait forever. The remaining drivers (ads, cbus, eip, knxnetip, opcua, s7) share the same hand-rolled pattern and should adopt DeliverResult in a follow-up. --- plc4go/internal/bacnetip/Reader.go | 19 ++-- .../internal/bacnetip/ReaderResultDelivery_test.go | 114 +++++++++++++++++++++ plc4go/internal/bacnetip/ReaderSegmentation.go | 5 +- plc4go/internal/bacnetip/Subscriber.go | 11 +- plc4go/internal/bacnetip/Writer.go | 13 +-- plc4go/internal/modbus/Reader.go | 33 +++--- plc4go/internal/modbus/Reader_test.go | 71 +++++++++++++ plc4go/internal/modbus/Writer.go | 29 ++---- plc4go/internal/modbus/Writer_test.go | 93 +++++++++++++++++ plc4go/internal/modbus/codec_stub_test.go | 59 +++++++++++ plc4go/spi/default/DefaultCodec.go | 25 ++++- .../default/DefaultCodecExpectationCleanup_test.go | 59 +++++++++++ plc4go/spi/utils/ResultDelivery.go | 39 +++++++ 13 files changed, 512 insertions(+), 58 deletions(-) diff --git a/plc4go/internal/bacnetip/Reader.go b/plc4go/internal/bacnetip/Reader.go index 5ff38b1952..5487ca8e46 100644 --- a/plc4go/internal/bacnetip/Reader.go +++ b/plc4go/internal/bacnetip/Reader.go @@ -34,6 +34,7 @@ import ( spiModel "github.com/apache/plc4x/plc4go/spi/model" "github.com/apache/plc4x/plc4go/spi/options" "github.com/apache/plc4x/plc4go/spi/transactions" + "github.com/apache/plc4x/plc4go/spi/utils" spiValues "github.com/apache/plc4x/plc4go/spi/values" ) @@ -69,7 +70,7 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) result := make(chan apiModel.PlcReadRequestResult, 1) m.wg.Go(func() { if len(readRequest.GetTagNames()) == 0 { - result <- spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.New("at least one field required")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.New("at least one field required"))) return } // create the service request @@ -188,32 +189,32 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) readResponse, err := m.ToPlc4xReadResponse(apdu, readRequest) if err != nil { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "Error decoding response"), - ) + )) return transaction.EndRequest() } - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, readResponse, nil, - ) + )) return transaction.EndRequest() }, func(err error) error { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "got timeout while waiting for response"), - ) + )) return transaction.EndRequest() }); err != nil { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "error sending message"), - ) + )) if err := transaction.FailRequest(errors.Errorf("timeout after %s", time.Second*1)); err != nil { m.log.Debug().Err(err).Msg("Error failing request") } diff --git a/plc4go/internal/bacnetip/ReaderResultDelivery_test.go b/plc4go/internal/bacnetip/ReaderResultDelivery_test.go new file mode 100644 index 0000000000..b576771da2 --- /dev/null +++ b/plc4go/internal/bacnetip/ReaderResultDelivery_test.go @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package bacnetip + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + apiModel "github.com/apache/plc4x/plc4go/pkg/api/model" + readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model" + "github.com/apache/plc4x/plc4go/spi" + "github.com/apache/plc4x/plc4go/spi/errors" + spiModel "github.com/apache/plc4x/plc4go/spi/model" + "github.com/apache/plc4x/plc4go/spi/testutils" + "github.com/apache/plc4x/plc4go/spi/transactions" +) + +// captureCodec is a minimal spi.MessageCodec stub that hands the callbacks +// passed to SendRequest back to the test and returns a configurable error. +type captureCodec struct { + sendRequestErr error + handlers chan capturedHandlers +} + +type capturedHandlers struct { + handleMessage spi.HandleMessage + handleError spi.HandleError +} + +func newCaptureCodec(sendRequestErr error) *captureCodec { + return &captureCodec{ + sendRequestErr: sendRequestErr, + handlers: make(chan capturedHandlers, 1), + } +} + +func (c *captureCodec) Connect(context.Context) error { return nil } +func (c *captureCodec) Disconnect() error { return nil } +func (c *captureCodec) IsRunning() bool { return true } +func (c *captureCodec) Send(context.Context, string, spi.Message) error { + return nil +} +func (c *captureCodec) Expect(context.Context, string, spi.AcceptsMessage, spi.HandleMessage, spi.HandleError) { +} +func (c *captureCodec) SendRequest(_ context.Context, _ string, _ spi.Message, _ spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) error { + c.handlers <- capturedHandlers{handleMessage: handleMessage, handleError: handleError} + return c.sendRequestErr +} +func (c *captureCodec) GetDefaultIncomingMessageChannel() chan spi.Message { return nil } + +// A caller whose context expired abandons the result channel without reading. +// The send-failure result then fills the single-slot buffer; a later +// error-handler invocation (expectation timeout, disconnect fan-out) must not +// block forever on the full channel — those blocked handlers pile up in the +// codec's WaitGroup and wedge Disconnect indefinitely. +func TestReader_lateErrorHandlerAfterFailedSendMustNotBlock(t *testing.T) { + codec := newCaptureCodec(errors.New("send failed: broken pipe")) + tm := transactions.NewRequestTransactionManager(1) + reader := NewReader(&InvokeIdGenerator{}, codec, tm) + + objType := readWriteModel.BACnetObjectType_ANALOG_INPUT + propId := readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE + tag := plcTag{ + ObjectId: objectId{ObjectIdType: &objType, ObjectIdInstance: 1}, + Properties: []property{{PropertyIdentifier: &propId}}, + } + request := spiModel.NewDefaultPlcReadRequest( + map[string]apiModel.PlcTag{"tag": tag}, []string{"tag"}, reader, nil) + + results := reader.Read(testutils.TestContext(t), request) + + var handlers capturedHandlers + select { + case handlers = <-codec.handlers: + case <-time.After(time.Second): + t.Fatal("SendRequest was never invoked") + } + + // Wait for the send-failure result to occupy the channel buffer; the + // abandoned caller never drains it. + require.Eventually(t, func() bool { return len(results) == 1 }, + time.Second, time.Millisecond) + + done := make(chan struct{}) + go func() { + defer close(done) + _ = handlers.handleError(errors.New("timeout")) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("late error handler blocked on the abandoned result channel") + } +} diff --git a/plc4go/internal/bacnetip/ReaderSegmentation.go b/plc4go/internal/bacnetip/ReaderSegmentation.go index 63f6b29173..e77ea331a1 100644 --- a/plc4go/internal/bacnetip/ReaderSegmentation.go +++ b/plc4go/internal/bacnetip/ReaderSegmentation.go @@ -28,6 +28,7 @@ import ( "github.com/apache/plc4x/plc4go/spi" "github.com/apache/plc4x/plc4go/spi/errors" spiModel "github.com/apache/plc4x/plc4go/spi/model" + "github.com/apache/plc4x/plc4go/spi/utils" ) // segmentWaitTimeout bounds how long we wait for each follow-up segment before @@ -102,12 +103,12 @@ func (m *Reader) reassembleSegmentedRead(ctx context.Context, readRequest apiMod m.failSegmentedRead(readRequest, result, errors.Wrap(err, "error decoding reassembled response")) return } - result <- spiModel.NewDefaultPlcReadRequestResult(readRequest, readResponse, nil) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, readResponse, nil)) } func (m *Reader) failSegmentedRead(readRequest apiModel.PlcReadRequest, result chan apiModel.PlcReadRequestResult, err error) { m.log.Debug().Err(err).Msg("segmented read failed") - result <- spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, err) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, err)) } // expectSegment registers a one-shot expectation for the next segmented diff --git a/plc4go/internal/bacnetip/Subscriber.go b/plc4go/internal/bacnetip/Subscriber.go index 3a447bbdcf..1bf90f11ba 100644 --- a/plc4go/internal/bacnetip/Subscriber.go +++ b/plc4go/internal/bacnetip/Subscriber.go @@ -34,6 +34,7 @@ import ( "github.com/apache/plc4x/plc4go/spi/errors" spiModel "github.com/apache/plc4x/plc4go/spi/model" "github.com/apache/plc4x/plc4go/spi/options" + "github.com/apache/plc4x/plc4go/spi/utils" spiValues "github.com/apache/plc4x/plc4go/spi/values" ) @@ -93,7 +94,7 @@ func (m *Subscriber) Subscribe(ctx context.Context, subscriptionRequest apiModel for _, tagName := range internalReq.GetTagNames() { if err := ctx.Err(); err != nil { - result <- spiModel.NewDefaultPlcSubscriptionRequestResult(subscriptionRequest, nil, err) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcSubscriptionRequestResult(subscriptionRequest, nil, err)) return } tag, ok := internalReq.GetTag(tagName).(BacNetPlcTag) @@ -123,7 +124,7 @@ func (m *Subscriber) Subscribe(ctx context.Context, subscriptionRequest apiModel } } - result <- spiModel.NewDefaultPlcSubscriptionRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcSubscriptionRequestResult( subscriptionRequest, spiModel.NewDefaultPlcSubscriptionResponse( subscriptionRequest, @@ -132,7 +133,7 @@ func (m *Subscriber) Subscribe(ctx context.Context, subscriptionRequest apiModel append(m._options, options.WithCustomLogger(m.log))..., ), nil, - ) + )) }) return result } @@ -145,7 +146,7 @@ func (m *Subscriber) Unsubscribe(ctx context.Context, unsubscriptionRequest apiM m.wg.Go(func() { req, ok := unsubscriptionRequest.(*spiModel.DefaultPlcUnsubscriptionRequest) if !ok { - result <- spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, errors.New("unsupported unsubscription request type")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, errors.New("unsupported unsubscription request type"))) return } for _, handle := range req.GetSubscriptionHandles() { @@ -157,7 +158,7 @@ func (m *Subscriber) Unsubscribe(ctx context.Context, unsubscriptionRequest apiM m.removeHandle(bh.subscriberProcessId) } // Build a response with OK for every handle the caller passed. - result <- spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, nil) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, nil)) }) return result } diff --git a/plc4go/internal/bacnetip/Writer.go b/plc4go/internal/bacnetip/Writer.go index b0080e7ecb..703304d485 100644 --- a/plc4go/internal/bacnetip/Writer.go +++ b/plc4go/internal/bacnetip/Writer.go @@ -34,6 +34,7 @@ import ( spiModel "github.com/apache/plc4x/plc4go/spi/model" "github.com/apache/plc4x/plc4go/spi/options" "github.com/apache/plc4x/plc4go/spi/transactions" + "github.com/apache/plc4x/plc4go/spi/utils" ) // Writer issues BACnet WriteProperty / WritePropertyMultiple confirmed-service @@ -72,18 +73,18 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques m.wg.Go(func() { defer func() { if r := recover(); r != nil { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", r, debug.Stack())) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", r, debug.Stack()))) } }() tagNames := writeRequest.GetTagNames() if len(tagNames) == 0 { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("at least one tag required")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("at least one tag required"))) return } serviceRequest, err := m.buildServiceRequest(writeRequest) if err != nil { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "Error building WriteProperty request")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "Error building WriteProperty request"))) return } @@ -113,14 +114,14 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques bvlc := message.(readWriteModel.BVLC) responseApdu := bvlc.(interface{ GetNpdu() readWriteModel.NPDU }).GetNpdu().GetApdu() writeResponse := m.toPlcWriteResponse(responseApdu, writeRequest) - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, writeResponse, nil) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, writeResponse, nil)) return transaction.EndRequest() }, func(err error) error { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "got timeout while waiting for write response")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "got timeout while waiting for write response"))) return transaction.EndRequest() }) if err != nil { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "error sending message")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "error sending message"))) if failErr := transaction.FailRequest(err); failErr != nil { m.log.Debug().Err(failErr).Msg("Error failing request") } diff --git a/plc4go/internal/modbus/Reader.go b/plc4go/internal/modbus/Reader.go index 767d322914..27c3020f9a 100644 --- a/plc4go/internal/modbus/Reader.go +++ b/plc4go/internal/modbus/Reader.go @@ -35,6 +35,7 @@ import ( "github.com/apache/plc4x/plc4go/spi/errors" spiModel "github.com/apache/plc4x/plc4go/spi/model" "github.com/apache/plc4x/plc4go/spi/options" + "github.com/apache/plc4x/plc4go/spi/utils" ) type Reader struct { @@ -67,11 +68,11 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) m.wg.Go(func() { defer func() { if err := recover(); err != nil { - result <- spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", err, debug.Stack())) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", err, debug.Stack()))) } }() if len(readRequest.GetTagNames()) != 1 { - result <- spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.New("modbus only supports single-item requests")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.New("modbus only supports single-item requests"))) m.log.Debug().Int("nTags", len(readRequest.GetTagNames())).Msg("modbus only supports single-item requests. Got nTags tags") return } @@ -80,11 +81,11 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) tag := readRequest.GetTag(tagName) modbusTagVar, err := castToModbusTagFromPlcTag(tag) if err != nil { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "invalid tag item type"), - ) + )) m.log.Debug().Type("tagType", tag).Msg("Invalid tag item type") return } @@ -101,18 +102,18 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) case HoldingRegister: pdu = readWriteModel.NewModbusPDUReadHoldingRegistersRequest(modbusTagVar.Address, numWords) case ExtendedRegister: - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.New("modbus currently doesn't support extended register requests"), - ) + )) return default: - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Errorf("unsupported tag type %x", modbusTagVar.TagType), - ) + )) m.log.Debug().Stringer("tagType", modbusTagVar.TagType).Msg("Unsupported tag type") return } @@ -144,33 +145,33 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) readResponse, err := m.ToPlc4xReadResponse(responseAdu, readRequest) if err != nil { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "Error decoding response"), - ) + )) // TODO: should we return the error here? return nil } - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, readResponse, nil, - ) + )) return nil }, func(err error) error { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "got timeout while waiting for response"), - ) + )) return nil }); err != nil { - result <- spiModel.NewDefaultPlcReadRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult( readRequest, nil, errors.Wrap(err, "error sending message"), - ) + )) } }) return result diff --git a/plc4go/internal/modbus/Reader_test.go b/plc4go/internal/modbus/Reader_test.go new file mode 100644 index 0000000000..d103bced9d --- /dev/null +++ b/plc4go/internal/modbus/Reader_test.go @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package modbus + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + apiModel "github.com/apache/plc4x/plc4go/pkg/api/model" + readWriteModel "github.com/apache/plc4x/plc4go/protocols/modbus/readwrite/model" + "github.com/apache/plc4x/plc4go/spi/errors" + spiModel "github.com/apache/plc4x/plc4go/spi/model" + "github.com/apache/plc4x/plc4go/spi/testutils" +) + +// A caller whose context expired abandons the result channel without reading. +// The send-failure result then fills the single-slot buffer; when the still +// registered expectation later times out, its error handler must not block +// forever on the full channel — those blocked handlers pile up in the codec's +// WaitGroup and wedge Disconnect indefinitely. +func TestReader_lateTimeoutAfterFailedSendMustNotBlock(t *testing.T) { + codec := newCaptureCodec(errors.New("send failed: broken pipe")) + reader := NewReader(1, codec) + tag := NewTag(HoldingRegister, 1, 1, readWriteModel.ModbusDataType_UINT) + request := spiModel.NewDefaultPlcReadRequest( + map[string]apiModel.PlcTag{"tag": tag}, []string{"tag"}, reader, nil) + + results := reader.Read(testutils.TestContext(t), request) + + var handlers capturedHandlers + select { + case handlers = <-codec.handlers: + case <-time.After(time.Second): + t.Fatal("SendRequest was never invoked") + } + + // Wait for the send-failure result to occupy the channel buffer; the + // abandoned caller never drains it. + require.Eventually(t, func() bool { return len(results) == 1 }, + time.Second, time.Millisecond) + + done := make(chan struct{}) + go func() { + defer close(done) + _ = handlers.handleError(errors.New("timeout")) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("late timeout handler blocked on the abandoned result channel") + } +} diff --git a/plc4go/internal/modbus/Writer.go b/plc4go/internal/modbus/Writer.go index d4a8ba9c91..17cb2fd179 100644 --- a/plc4go/internal/modbus/Writer.go +++ b/plc4go/internal/modbus/Writer.go @@ -33,6 +33,7 @@ import ( "github.com/apache/plc4x/plc4go/spi/errors" spiModel "github.com/apache/plc4x/plc4go/spi/model" "github.com/apache/plc4x/plc4go/spi/options" + "github.com/apache/plc4x/plc4go/spi/utils" ) type Writer struct { @@ -61,7 +62,7 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques m.wg.Go(func() { // If we are requesting only one tag, use a if len(writeRequest.GetTagNames()) != 1 { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("modbus only supports single-item requests")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("modbus only supports single-item requests"))) return } tagName := writeRequest.GetTagNames()[0] @@ -70,7 +71,7 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques tag := writeRequest.GetTag(tagName) modbusTag, err := castToModbusTagFromPlcTag(tag) if err != nil { - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "invalid tag item type")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "invalid tag item type"))) return } @@ -78,11 +79,11 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques value := writeRequest.GetValue(tagName) data, err := readWriteModel.DataItemSerialize(value, modbusTag.Datatype, modbusTag.Quantity, true) if err != nil { - result <- spiModel.NewDefaultPlcWriteRequestResult( + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult( writeRequest, nil, errors.Wrap(err, "error serializing value"), - ) + )) return } @@ -102,10 +103,10 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques numWords, data) case ExtendedRegister: - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("modbus currently doesn't support extended register requests")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("modbus currently doesn't support extended register requests"))) return default: - result <- spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("unsupported tag type")) + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.New("unsupported tag type"))) return } @@ -131,25 +132,17 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques readResponse, err := m.ToPlc4xWriteResponse(requestAdu, responseAdu, writeRequest) if err != nil { - result <- &spiModel.DefaultPlcWriteRequestResult{ - Request: writeRequest, - Err: errors.Wrap(err, "Error decoding response"), - } + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "Error decoding response"))) } else { - result <- &spiModel.DefaultPlcWriteRequestResult{ - Request: writeRequest, - Response: readResponse, - } + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, readResponse, nil)) } return nil }, func(err error) error { - result <- &spiModel.DefaultPlcWriteRequestResult{ - Request: writeRequest, - Err: errors.New("got timeout while waiting for response"), - } + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "got timeout while waiting for response"))) return nil }); err != nil { m.log.Debug().Err(err).Msg("error sending message") + utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrap(err, "error sending message"))) } }) return result diff --git a/plc4go/internal/modbus/Writer_test.go b/plc4go/internal/modbus/Writer_test.go new file mode 100644 index 0000000000..7922ce125f --- /dev/null +++ b/plc4go/internal/modbus/Writer_test.go @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package modbus + +import ( + "testing" + "time" + + apiModel "github.com/apache/plc4x/plc4go/pkg/api/model" + apiValues "github.com/apache/plc4x/plc4go/pkg/api/values" + readWriteModel "github.com/apache/plc4x/plc4go/protocols/modbus/readwrite/model" + "github.com/apache/plc4x/plc4go/spi/errors" + spiModel "github.com/apache/plc4x/plc4go/spi/model" + "github.com/apache/plc4x/plc4go/spi/testutils" + spiValues "github.com/apache/plc4x/plc4go/spi/values" +) + +func testWriteRequest(t *testing.T, writer *Writer) apiModel.PlcWriteRequest { + t.Helper() + tag := NewTag(HoldingRegister, 1, 1, readWriteModel.ModbusDataType_UINT) + return spiModel.NewDefaultPlcWriteRequest( + map[string]apiModel.PlcTag{"tag": tag}, + []string{"tag"}, + map[string]apiValues.PlcValue{"tag": spiValues.NewPlcUINT(42)}, + writer, + nil, + ) +} + +// A failed send must deliver an error result to the caller instead of only +// logging it — a caller without its own deadline otherwise waits forever on +// a channel that never receives anything. +func TestWriter_failedSendDeliversErrorResult(t *testing.T) { + codec := newCaptureCodec(errors.New("send failed: broken pipe")) + writer := NewWriter(1, codec) + + results := writer.Write(testutils.TestContext(t), testWriteRequest(t, writer)) + + select { + case result := <-results: + if result.GetErr() == nil { + t.Fatal("expected an error result for the failed send") + } + case <-time.After(2 * time.Second): + t.Fatal("no result delivered for a failed send") + } +} + +// When the caller has abandoned the result channel, a duplicate error-handler +// invocation (e.g. expectation timeout racing a handled message, or the +// disconnect fan-out) must not block forever on the full single-slot buffer. +func TestWriter_duplicateErrorHandlerMustNotBlock(t *testing.T) { + codec := newCaptureCodec(nil) + writer := NewWriter(1, codec) + + _ = writer.Write(testutils.TestContext(t), testWriteRequest(t, writer)) + + var handlers capturedHandlers + select { + case handlers = <-codec.handlers: + case <-time.After(time.Second): + t.Fatal("SendRequest was never invoked") + } + + done := make(chan struct{}) + go func() { + defer close(done) + _ = handlers.handleError(errors.New("timeout")) + _ = handlers.handleError(errors.New("disconnected")) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("duplicate error handler blocked on the abandoned result channel") + } +} diff --git a/plc4go/internal/modbus/codec_stub_test.go b/plc4go/internal/modbus/codec_stub_test.go new file mode 100644 index 0000000000..756b305ff7 --- /dev/null +++ b/plc4go/internal/modbus/codec_stub_test.go @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package modbus + +import ( + "context" + + "github.com/apache/plc4x/plc4go/spi" +) + +// captureCodec is a minimal spi.MessageCodec stub that hands the callbacks +// passed to SendRequest back to the test and returns a configurable error. +type captureCodec struct { + sendRequestErr error + handlers chan capturedHandlers +} + +type capturedHandlers struct { + handleMessage spi.HandleMessage + handleError spi.HandleError +} + +func newCaptureCodec(sendRequestErr error) *captureCodec { + return &captureCodec{ + sendRequestErr: sendRequestErr, + handlers: make(chan capturedHandlers, 1), + } +} + +func (c *captureCodec) Connect(context.Context) error { return nil } +func (c *captureCodec) Disconnect() error { return nil } +func (c *captureCodec) IsRunning() bool { return true } +func (c *captureCodec) Send(context.Context, string, spi.Message) error { + return nil +} +func (c *captureCodec) Expect(context.Context, string, spi.AcceptsMessage, spi.HandleMessage, spi.HandleError) { +} +func (c *captureCodec) SendRequest(_ context.Context, _ string, _ spi.Message, _ spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) error { + c.handlers <- capturedHandlers{handleMessage: handleMessage, handleError: handleError} + return c.sendRequestErr +} +func (c *captureCodec) GetDefaultIncomingMessageChannel() chan spi.Message { return nil } diff --git a/plc4go/spi/default/DefaultCodec.go b/plc4go/spi/default/DefaultCodec.go index e60f92a97d..be59bcb655 100644 --- a/plc4go/spi/default/DefaultCodec.go +++ b/plc4go/spi/default/DefaultCodec.go @@ -223,6 +223,12 @@ func (m *defaultCodec) IsRunning() bool { } func (m *defaultCodec) Expect(ctx context.Context, interactionInfo string, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) { + m.expect(ctx, interactionInfo, acceptsMessage, handleMessage, handleError) +} + +// expect is the implementation of Expect which additionally hands back the +// registered expectation so internal callers can remove it again. +func (m *defaultCodec) expect(ctx context.Context, interactionInfo string, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) spi.Expectation { m.expectationsChangeMutex.Lock() defer m.expectationsChangeMutex.Unlock() ttl := m.receiveTimeout @@ -240,15 +246,30 @@ func (m *defaultCodec) Expect(ctx context.Context, interactionInfo string, accep case m.notifyReceiveWorker <- struct{}{}: default: } + return expectation +} + +func (m *defaultCodec) removeExpectation(expectation spi.Expectation) { + m.expectationsChangeMutex.Lock() + defer m.expectationsChangeMutex.Unlock() + m.expectations = slices.DeleteFunc(m.expectations, func(candidate spi.Expectation) bool { + return candidate == expectation + }) } func (m *defaultCodec) SendRequest(ctx context.Context, interactionInfo string, message spi.Message, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) error { if err := ctx.Err(); err != nil { return errors.Wrap(err, "Not sending message as context is aborted") } - m.Expect(ctx, interactionInfo, acceptsMessage, handleMessage, handleError) // We register the expectation first to avoid getting a response between sending and adding the expect + expectation := m.expect(ctx, interactionInfo, acceptsMessage, handleMessage, handleError) // We register the expectation first to avoid getting a response between sending and adding the expect m.log.Trace().Str("interactionInfo", interactionInfo).Msg("Sending request") - return m.Send(ctx, interactionInfo, message) + if err := m.Send(ctx, interactionInfo, message); err != nil { + // The caller receives the send error directly; leaving the expectation + // registered would fire the error handler a second time on timeout. + m.removeExpectation(expectation) + return err + } + return nil } func (m *defaultCodec) TimeoutExpectations(now time.Time) time.Duration { diff --git a/plc4go/spi/default/DefaultCodecExpectationCleanup_test.go b/plc4go/spi/default/DefaultCodecExpectationCleanup_test.go new file mode 100644 index 0000000000..02e75937e3 --- /dev/null +++ b/plc4go/spi/default/DefaultCodecExpectationCleanup_test.go @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package _default + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + + "github.com/apache/plc4x/plc4go/spi" + "github.com/apache/plc4x/plc4go/spi/errors" + "github.com/apache/plc4x/plc4go/spi/testutils" +) + +// A SendRequest whose Send fails must not leave its expectation registered: +// the caller already receives the send error, so a later expectation timeout +// would fire the error handler a second time for the same request. +func Test_defaultCodec_SendRequest_failedSendRemovesExpectation(t *testing.T) { + requirements := NewMockDefaultCodecRequirements(t) + requirements.EXPECT().Send(mock.Anything, mock.Anything, mock.Anything).Return(errors.New("nope")) + m := &defaultCodec{ + DefaultCodecRequirements: requirements, + notifyExpireWorker: make(chan struct{}, 100), + notifyReceiveWorker: make(chan struct{}, 100), + log: testutils.ProduceTestingLogger(t), + } + + err := m.SendRequest( + testutils.TestContext(t), + t.Name(), + nil, + func(spi.Message) bool { return true }, + func(spi.Message) error { return nil }, + func(error) error { return nil }, + ) + assert.Error(t, err) + + m.expectationsChangeMutex.RLock() + defer m.expectationsChangeMutex.RUnlock() + assert.Empty(t, m.expectations, "failed send left its expectation registered") +} diff --git a/plc4go/spi/utils/ResultDelivery.go b/plc4go/spi/utils/ResultDelivery.go new file mode 100644 index 0000000000..51148da081 --- /dev/null +++ b/plc4go/spi/utils/ResultDelivery.go @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package utils + +import ( + "github.com/rs/zerolog" +) + +// DeliverResult hands a result to the caller without ever blocking. The result +// channels have capacity 1 and their consumer reads at most one result (a +// caller whose context expired reads none), so a surplus result — a late +// expectation timeout after a send failure, a duplicate error-handler +// invocation, or any result for an abandoned request — must be dropped instead +// of pinning the sending goroutine forever. Such pinned goroutines accumulate +// in the codec's WaitGroup and wedge Disconnect indefinitely. +func DeliverResult[T any](log zerolog.Logger, results chan<- T, result T) { + select { + case results <- result: + default: + log.Warn().Msg("dropping result: result channel is full and nobody is reading it anymore") + } +}
