You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2024/09/04 10:37:02 UTC
(plc4x) 01/06: feat(plc4go/bacnetip): add wait groups to ensure async shutdowns
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 3e0171b7a31cec5c5dc649532f155e52ac107a4c
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Sep 3 21:07:59 2024 +0200
feat(plc4go/bacnetip): add wait groups to ensure async shutdowns
---
.../internal/bacnetip/app_ApplicationIOController.go | 20 +++++++++++++++-----
.../internal/bacnetip/app_BIPForeignApplication.go | 6 ++++++
plc4go/internal/bacnetip/app_BIPSimpleApplication.go | 6 ++++++
plc4go/internal/bacnetip/iocb_IOCB.go | 14 ++++++++++++++
plc4go/internal/bacnetip/iocb_IOQController.go | 7 +++++++
plc4go/internal/bacnetip/iocb_IOQueue.go | 14 ++++++++++++++
.../internal/bacnetip/tests/test_service/helpers.go | 10 ++++++++++
.../bacnetip/tests/test_service/test_cov_test.go | 3 +++
plc4go/internal/bacnetip/udp_UDPDirector.go | 15 +++++++++++++++
9 files changed, 90 insertions(+), 5 deletions(-)
diff --git a/plc4go/internal/bacnetip/app_ApplicationIOController.go b/plc4go/internal/bacnetip/app_ApplicationIOController.go
index 785e4a24f3..8b0ea299c0 100644
--- a/plc4go/internal/bacnetip/app_ApplicationIOController.go
+++ b/plc4go/internal/bacnetip/app_ApplicationIOController.go
@@ -30,7 +30,7 @@ type ApplicationIOController struct {
*IOController
*Application
- queueByAddress map[string]SieveQueue
+ queueByAddress map[string]*SieveQueue
// pass through args
argDeviceInfoCache *DeviceInfoCache
@@ -42,7 +42,7 @@ type ApplicationIOController struct {
func NewApplicationIOController(localLog zerolog.Logger, localDevice *LocalDeviceObject, opts ...func(controller *ApplicationIOController)) (*ApplicationIOController, error) {
a := &ApplicationIOController{
// queues for each address
- queueByAddress: make(map[string]SieveQueue),
+ queueByAddress: make(map[string]*SieveQueue),
log: localLog,
}
for _, opt := range opts {
@@ -86,10 +86,10 @@ func (a *ApplicationIOController) ProcessIO(iocb _IOCB) error {
queue, ok := a.queueByAddress[destinationAddress.String()]
if !ok {
newQueue, _ := NewSieveQueue(a.log, a._AppRequest, destinationAddress)
- queue = *newQueue
+ queue = newQueue
a.queueByAddress[destinationAddress.String()] = queue
}
- a.log.Debug().Stringer("queue", &queue).Msg("working with queue")
+ a.log.Debug().Stringer("queue", queue).Msg("working with queue")
// ask the queue to process the request
return queue.RequestIO(iocb)
@@ -107,7 +107,7 @@ func (a *ApplicationIOController) _AppComplete(address *Address, apdu PDU) error
a.log.Debug().Stringer("address", address).Msg("no queue for")
return nil
}
- a.log.Debug().Stringer("queue", &queue).Msg("working with queue")
+ a.log.Debug().Stringer("queue", queue).Msg("working with queue")
// make sure it has an active iocb
if queue.activeIOCB == nil {
@@ -175,3 +175,13 @@ func (a *ApplicationIOController) Confirmation(args Args, kwargs KWArgs) error {
// this is an ack, error, reject or abort
return a._AppComplete(apdu.GetPDUSource(), apdu)
}
+
+func (a *ApplicationIOController) Close() error {
+ for addr, queue := range a.queueByAddress {
+ a.log.Debug().Str("addr", addr).Msg("Closing")
+ if err := queue.Close(); err != nil {
+ a.log.Warn().Str("addr", addr).Err(err).Stringer("queue", queue).Msg("error closing")
+ }
+ }
+ return nil
+}
diff --git a/plc4go/internal/bacnetip/app_BIPForeignApplication.go b/plc4go/internal/bacnetip/app_BIPForeignApplication.go
index daba87c177..eb8111e992 100644
--- a/plc4go/internal/bacnetip/app_BIPForeignApplication.go
+++ b/plc4go/internal/bacnetip/app_BIPForeignApplication.go
@@ -125,6 +125,12 @@ func NewBIPForeignApplication(localLog zerolog.Logger, localDevice *LocalDeviceO
func (b *BIPForeignApplication) Close() error {
b.log.Debug().Msg("close socket")
+ if b.ApplicationIOController != nil {
+ if err := b.ApplicationIOController.Close(); err != nil {
+ b.log.Warn().Err(err).Msg("error closing applicationIOController")
+ }
+ }
+
// pass to the multiplexer, then down to the sockets
return b.mux.Close()
}
diff --git a/plc4go/internal/bacnetip/app_BIPSimpleApplication.go b/plc4go/internal/bacnetip/app_BIPSimpleApplication.go
index 24287eb01c..37722db18c 100644
--- a/plc4go/internal/bacnetip/app_BIPSimpleApplication.go
+++ b/plc4go/internal/bacnetip/app_BIPSimpleApplication.go
@@ -126,6 +126,12 @@ func NewBIPSimpleApplication(localLog zerolog.Logger, localDevice *LocalDeviceOb
func (b *BIPSimpleApplication) Close() error {
b.log.Debug().Msg("close socket")
+ if b.ApplicationIOController != nil {
+ if err := b.ApplicationIOController.Close(); err != nil {
+ b.log.Warn().Err(err).Msg("error closing applicationIOController")
+ }
+ }
+
// pass to the multiplexer, then down to the sockets
return b.mux.Close()
}
diff --git a/plc4go/internal/bacnetip/iocb_IOCB.go b/plc4go/internal/bacnetip/iocb_IOCB.go
index 0bd45de4bd..f97c9e2e85 100644
--- a/plc4go/internal/bacnetip/iocb_IOCB.go
+++ b/plc4go/internal/bacnetip/iocb_IOCB.go
@@ -91,6 +91,8 @@ type IOCB struct {
ioTimoutCancel chan any
priority int
+ wg sync.WaitGroup
+
log zerolog.Logger `ignore:"true"`
}
@@ -236,7 +238,9 @@ func (i *IOCB) SetTimeout(delay time.Duration) {
now := GetTaskManagerTime()
i.ioTimeout = time.NewTimer(delay)
i.ioTimoutCancel = make(chan any)
+ i.wg.Add(1)
go func() {
+ defer i.wg.Done()
select {
case timeout := <-i.ioTimeout.C:
_ = i.Abort(utils.NewTimeoutError(now.Sub(timeout)))
@@ -281,3 +285,13 @@ func (i *IOCB) getPriority() int {
func (i *IOCB) clearQueue() {
i.ioQueue = nil
}
+
+func (i *IOCB) Close() error { // TODO: ensure this is getting called
+ i.log.Debug().Msg("IOCB closing")
+ defer func() {
+ i.log.Debug().Msg("waiting for running tasks to finnish")
+ i.wg.Wait()
+ i.log.Debug().Msg("waiting done")
+ }()
+ return nil
+}
diff --git a/plc4go/internal/bacnetip/iocb_IOQController.go b/plc4go/internal/bacnetip/iocb_IOQController.go
index 06bd3e80f9..ff63bc09a8 100644
--- a/plc4go/internal/bacnetip/iocb_IOQController.go
+++ b/plc4go/internal/bacnetip/iocb_IOQController.go
@@ -278,3 +278,10 @@ func (i *IOQController) _waitTrigger(_ Args, _ KWArgs) error {
// look for more to do
return i._trigger(NoArgs, NoKWArgs)
}
+
+func (i *IOQController) Close() error {
+ if i.ioQueue != nil {
+ return i.ioQueue.Close()
+ }
+ return nil
+}
diff --git a/plc4go/internal/bacnetip/iocb_IOQueue.go b/plc4go/internal/bacnetip/iocb_IOQueue.go
index b265778cfa..e779221cde 100644
--- a/plc4go/internal/bacnetip/iocb_IOQueue.go
+++ b/plc4go/internal/bacnetip/iocb_IOQueue.go
@@ -36,6 +36,8 @@ type IOQueue struct {
notEmpty sync.Cond
queue PriorityQueue[int, _IOCB]
+ wg sync.WaitGroup
+
log zerolog.Logger `ignore:"true"`
}
@@ -84,7 +86,9 @@ func (i *IOQueue) Get(block bool, delay *time.Duration) (_IOCB, error) {
if len(i.queue) == 0 {
if delay != nil {
gotSomething := make(chan any)
+ i.wg.Add(1)
go func() {
+ defer i.wg.Done()
i.notEmpty.Wait()
close(gotSomething)
}()
@@ -143,3 +147,13 @@ func (i *IOQueue) Abort(err error) {
// the queue is now empty, clear the event
i.notEmpty.Broadcast()
}
+
+func (i *IOQueue) Close() error {
+ i.log.Debug().Msg("IOQueue closing")
+ defer func() {
+ i.log.Debug().Msg("waiting for running tasks to finnish")
+ i.wg.Wait()
+ i.log.Debug().Msg("waiting done")
+ }()
+ return nil
+}
diff --git a/plc4go/internal/bacnetip/tests/test_service/helpers.go b/plc4go/internal/bacnetip/tests/test_service/helpers.go
index 16783492d6..3120d78067 100644
--- a/plc4go/internal/bacnetip/tests/test_service/helpers.go
+++ b/plc4go/internal/bacnetip/tests/test_service/helpers.go
@@ -157,6 +157,16 @@ func (a *ApplicationNetwork) _debug(format string, args bacnetip.Args) {
a.log.Debug().Msgf(format, args)
}
+func (a *ApplicationNetwork) Close() error {
+ if err := a.td.Close(); err != nil {
+ a.log.Warn().Err(err).Msg("error closing td")
+ }
+ if err := a.iut.Close(); err != nil {
+ a.log.Warn().Err(err).Msg("error closing iut")
+ }
+ return nil
+}
+
type SnifferNode struct {
bacnetip.Client
diff --git a/plc4go/internal/bacnetip/tests/test_service/test_cov_test.go b/plc4go/internal/bacnetip/tests/test_service/test_cov_test.go
index 119d970c7a..d9887beee2 100644
--- a/plc4go/internal/bacnetip/tests/test_service/test_cov_test.go
+++ b/plc4go/internal/bacnetip/tests/test_service/test_cov_test.go
@@ -39,6 +39,9 @@ func TestBasic(t *testing.T) {
// create a network
anet, err := NewApplicationNetwork(testingLogger)
require.NoError(t, err)
+ t.Cleanup(func() {
+ require.NoError(t, anet.Close())
+ })
// add the service capability
anet.iut.AddCapability(new(service.ChangeOfValuesServices))
diff --git a/plc4go/internal/bacnetip/udp_UDPDirector.go b/plc4go/internal/bacnetip/udp_UDPDirector.go
index e2adfd501f..6b39922d07 100644
--- a/plc4go/internal/bacnetip/udp_UDPDirector.go
+++ b/plc4go/internal/bacnetip/udp_UDPDirector.go
@@ -23,6 +23,7 @@ import (
"context"
"fmt"
"net"
+ "sync"
"github.com/libp2p/go-reuseport"
"github.com/pkg/errors"
@@ -46,6 +47,8 @@ type UDPDirector struct {
peers map[string]*UDPActor
running bool
+ wg sync.WaitGroup
+
passLogToModel bool
log zerolog.Logger
}
@@ -99,7 +102,9 @@ func NewUDPDirector(localLog zerolog.Logger, address AddressTuple[string, uint16
}
d.running = true
+ d.wg.Add(1)
go func() {
+ defer d.wg.Done()
for d.running {
d.handleRead()
}
@@ -107,7 +112,9 @@ func NewUDPDirector(localLog zerolog.Logger, address AddressTuple[string, uint16
// create the request queue
d.request = make(chan PDU)
+ d.wg.Add(1)
go func() {
+ defer d.wg.Done()
for d.running {
pdu := <-d.request
serialize, err := pdu.GetRootMessage().Serialize()
@@ -184,6 +191,12 @@ func (d *UDPDirector) ActorError(actor *UDPActor, err error) {
}
func (d *UDPDirector) Close() error {
+ d.log.Debug().Msg("UDPDirector closing")
+ defer func() {
+ d.log.Debug().Msg("waiting for running tasks to finnish")
+ d.wg.Wait()
+ d.log.Debug().Msg("waiting done")
+ }()
d.running = false
return d.udpConn.Close()
}
@@ -222,7 +235,9 @@ func (d *UDPDirector) handleRead() {
}
pdu := NewPDU(bvlc, WithPDUSource(saddr), WithPDUDestination(daddr))
// send the _PDU up to the client
+ d.wg.Add(1)
go func() {
+ defer d.wg.Done()
if err := d._response(pdu); err != nil {
d.log.Debug().Err(err).Msg("errored")
}