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 2023/07/12 14:49:37 UTC
[plc4x] branch develop updated (b2512192af -> 8968760a27)
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
from b2512192af test(plc4go/spi): add test for TimeoutExpectations in DefaultCodec ensuring it works properly
new 7d03458266 refactor(plc4go/spi): avoid issues when using executor and logging
new 8968760a27 test(plc4go/cbus): improve assertion
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
plc4go/internal/cbus/Browser_test.go | 14 +++-----------
plc4go/spi/pool/WorkerPool.go | 30 ++++++------------------------
plc4go/spi/pool/dynamicExecutor.go | 13 ++++---------
plc4go/spi/pool/dynamicExecutor_test.go | 2 +-
plc4go/spi/pool/executor.go | 14 ++++++++++----
plc4go/spi/pool/executor_test.go | 14 ++++++++------
plc4go/spi/pool/worker.go | 22 ++++++++++++++++++----
7 files changed, 50 insertions(+), 59 deletions(-)
[plc4x] 01/02: refactor(plc4go/spi): avoid issues when using executor and logging
Posted by sr...@apache.org.
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 7d0345826641f97d4f49454635fc3a24946a9345
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Jul 12 13:28:12 2023 +0200
refactor(plc4go/spi): avoid issues when using executor and logging
---
plc4go/spi/pool/WorkerPool.go | 30 ++++++------------------------
plc4go/spi/pool/dynamicExecutor.go | 13 ++++---------
plc4go/spi/pool/dynamicExecutor_test.go | 2 +-
plc4go/spi/pool/executor.go | 14 ++++++++++----
plc4go/spi/pool/executor_test.go | 14 ++++++++------
plc4go/spi/pool/worker.go | 22 ++++++++++++++++++----
6 files changed, 47 insertions(+), 48 deletions(-)
diff --git a/plc4go/spi/pool/WorkerPool.go b/plc4go/spi/pool/WorkerPool.go
index b5d781fb82..7611ce242e 100644
--- a/plc4go/spi/pool/WorkerPool.go
+++ b/plc4go/spi/pool/WorkerPool.go
@@ -21,11 +21,10 @@ package pool
import (
"context"
-
- "github.com/apache/plc4x/plc4go/spi/options"
-
"io"
"time"
+
+ "github.com/apache/plc4x/plc4go/spi/options"
)
type Runnable func()
@@ -44,21 +43,9 @@ type Executor interface {
}
func NewFixedSizeExecutor(numberOfWorkers, queueDepth int, _options ...options.WithOption) Executor {
- workers := make([]*worker, numberOfWorkers)
customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
- for i := 0; i < numberOfWorkers; i++ {
- w := worker{
- id: i,
- log: customLogger,
- }
- w.lastReceived.Store(time.Time{})
- workers[i] = &w
- }
- _executor := newExecutor(queueDepth, workers, customLogger)
+ _executor := newExecutor(queueDepth, numberOfWorkers, customLogger)
_executor.traceWorkers, _ = options.ExtractTracerWorkers(_options...)
- for i := 0; i < numberOfWorkers; i++ {
- workers[i].executor = _executor
- }
return _executor
}
@@ -67,13 +54,8 @@ func NewDynamicExecutor(maxNumberOfWorkers, queueDepth int, _options ...options.
_executor := newDynamicExecutor(queueDepth, maxNumberOfWorkers, customLogger)
_executor.traceWorkers, _ = options.ExtractTracerWorkers(_options...)
// We spawn one initial worker
- w := worker{
- id: 0,
- interrupter: make(chan struct{}, 1),
- executor: _executor,
- log: customLogger,
- }
- w.lastReceived.Store(time.Now())
- _executor.worker = append(_executor.worker, &w)
+ w := newWorker(customLogger, 0, _executor)
+ w.lastReceived.Store(time.Now()) // We store the current timestamp so the worker isn't cut of instantly by the worker killer
+ _executor.worker = append(_executor.worker, w)
return _executor
}
diff --git a/plc4go/spi/pool/dynamicExecutor.go b/plc4go/spi/pool/dynamicExecutor.go
index 630cabee08..72e3a0978a 100644
--- a/plc4go/spi/pool/dynamicExecutor.go
+++ b/plc4go/spi/pool/dynamicExecutor.go
@@ -46,7 +46,7 @@ type dynamicExecutor struct {
func newDynamicExecutor(queueDepth, maxNumberOfWorkers int, log zerolog.Logger) *dynamicExecutor {
return &dynamicExecutor{
- executor: newExecutor(queueDepth, make([]*worker, 0), log),
+ executor: newExecutor(queueDepth, 0, log),
maxNumberOfWorkers: maxNumberOfWorkers,
}
}
@@ -88,15 +88,10 @@ func (e *dynamicExecutor) Start() {
workerLog.Debug().Msgf("Checking if numberOfItemsInQueue(%d) > numberOfWorkers(%d) && numberOfWorkers(%d) < maxNumberOfWorkers(%d)", numberOfItemsInQueue, numberOfWorkers, numberOfWorkers, e.maxNumberOfWorkers)
if numberOfItemsInQueue > numberOfWorkers && numberOfWorkers < e.maxNumberOfWorkers {
workerLog.Trace().Msg("spawning new worker")
- _worker := &worker{
- id: numberOfWorkers - 1,
- interrupter: make(chan struct{}, 1),
- executor: e,
- log: e.log,
- }
- _worker.lastReceived.Store(time.Now())
+ workerId := numberOfWorkers - 1
+ _worker := newWorker(e.log, workerId, e)
+ _worker.lastReceived.Store(time.Now()) // We store the current timestamp so the worker isn't cut of instantly by the worker killer
e.worker = append(e.worker, _worker)
- _worker.initialize()
workerLog.Info().Int("Worker id", _worker.id).Msg("spawning")
_worker.start()
e.currentNumberOfWorkers.Add(1)
diff --git a/plc4go/spi/pool/dynamicExecutor_test.go b/plc4go/spi/pool/dynamicExecutor_test.go
index 482c9e729d..536d1544e9 100644
--- a/plc4go/spi/pool/dynamicExecutor_test.go
+++ b/plc4go/spi/pool/dynamicExecutor_test.go
@@ -43,7 +43,7 @@ func Test_newDynamicExecutor(t *testing.T) {
{
name: "just create it",
want: &dynamicExecutor{
- executor: newExecutor(0, make([]*worker, 0), zerolog.Logger{}),
+ executor: newExecutor(0, 0, zerolog.Logger{}),
},
manipulator: func(t *testing.T, want *dynamicExecutor, got *dynamicExecutor) {
assert.NotNil(t, got.workItems)
diff --git a/plc4go/spi/pool/executor.go b/plc4go/spi/pool/executor.go
index e3c83917a2..e2f34e6606 100644
--- a/plc4go/spi/pool/executor.go
+++ b/plc4go/spi/pool/executor.go
@@ -43,12 +43,18 @@ type executor struct {
log zerolog.Logger `ignore:"true"`
}
-func newExecutor(queueDepth int, workers []*worker, log zerolog.Logger) *executor {
- return &executor{
+func newExecutor(queueDepth int, numberOfInitialWorkers int, customLogger zerolog.Logger) *executor {
+ e := &executor{
workItems: make(chan workItem, queueDepth),
- worker: workers,
- log: log,
+ log: customLogger,
}
+ workers := make([]*worker, numberOfInitialWorkers)
+ for i := 0; i < numberOfInitialWorkers; i++ {
+ w := newWorker(customLogger, i, e)
+ workers[i] = w
+ }
+ e.worker = workers
+ return e
}
func (e *executor) isTraceWorkers() bool {
diff --git a/plc4go/spi/pool/executor_test.go b/plc4go/spi/pool/executor_test.go
index 3f575ca75f..5e56eb92a1 100644
--- a/plc4go/spi/pool/executor_test.go
+++ b/plc4go/spi/pool/executor_test.go
@@ -33,9 +33,9 @@ import (
func Test_newExecutor(t *testing.T) {
type args struct {
- queueDepth int
- workers []*worker
- log zerolog.Logger
+ queueDepth int
+ numberOfInitialWorkers int
+ log zerolog.Logger
}
tests := []struct {
name string
@@ -45,7 +45,9 @@ func Test_newExecutor(t *testing.T) {
}{
{
name: "just create it",
- want: &executor{},
+ want: &executor{
+ worker: []*worker{},
+ },
manipulator: func(t *testing.T, want *executor, got *executor) {
assert.NotNil(t, got.workItems)
want.workItems = got.workItems
@@ -54,12 +56,12 @@ func Test_newExecutor(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got := newExecutor(tt.args.queueDepth, tt.args.workers, tt.args.log)
+ got := newExecutor(tt.args.queueDepth, tt.args.numberOfInitialWorkers, tt.args.log)
want := tt.want
if tt.manipulator != nil {
tt.manipulator(t, want, got)
}
- assert.Equalf(t, want, got, "newExecutor(%v, %v, %v)", tt.args.queueDepth, tt.args.workers, tt.args.log)
+ assert.Equalf(t, want, got, "newExecutor(%v, %v, %v)", tt.args.queueDepth, tt.args.numberOfInitialWorkers, tt.args.log)
})
}
}
diff --git a/plc4go/spi/pool/worker.go b/plc4go/spi/pool/worker.go
index f58e8dcbde..e61a65659e 100644
--- a/plc4go/spi/pool/worker.go
+++ b/plc4go/spi/pool/worker.go
@@ -49,10 +49,24 @@ type worker struct {
log zerolog.Logger `ignore:"true"`
}
+func newWorker(localLog zerolog.Logger, workerId int, executor interface {
+ isTraceWorkers() bool
+ getWorksItems() chan workItem
+ getWorkerWaitGroup() *sync.WaitGroup
+}) *worker {
+ w := &worker{
+ id: workerId,
+ executor: executor,
+ log: localLog.With().Int("workerId", workerId).Logger(),
+ }
+ w.initialize()
+ return w
+}
+
func (w *worker) initialize() {
w.stateChange.Lock()
defer w.stateChange.Unlock()
- w.lastReceived.Store(time.Now())
+ w.lastReceived.Store(time.Time{})
w.running.Store(false)
w.shutdown.Store(false)
w.interrupted.Store(false)
@@ -114,15 +128,15 @@ func (w *worker) work() {
select {
case _workItem := <-w.executor.getWorksItems():
w.lastReceived.Store(time.Now())
- workerLog.Debug().Msgf("Got work item %v", _workItem)
+ workerLog.Debug().Msgf("Got work item\n%s", &_workItem)
if _workItem.completionFuture.cancelRequested.Load() || (w.shutdown.Load() && w.interrupted.Load()) {
workerLog.Debug().Msg("We need to stop")
// TODO: do we need to complete with a error?
} else {
- workerLog.Debug().Msgf("Running work item %v", _workItem)
+ workerLog.Debug().Msgf("Running work item\n%s", &_workItem)
_workItem.runnable()
_workItem.completionFuture.complete()
- workerLog.Debug().Msgf("work item %v completed", _workItem)
+ workerLog.Debug().Msgf("work item completed\n%s", &_workItem)
}
case <-w.interrupter:
workerLog.Debug().Msg("We got interrupted")
[plc4x] 02/02: test(plc4go/cbus): improve assertion
Posted by sr...@apache.org.
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 8968760a27b0f99315c39f6c5af9bb49ddc02f02
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Jul 12 16:18:02 2023 +0200
test(plc4go/cbus): improve assertion
---
plc4go/internal/cbus/Browser_test.go | 14 +++-----------
1 file changed, 3 insertions(+), 11 deletions(-)
diff --git a/plc4go/internal/cbus/Browser_test.go b/plc4go/internal/cbus/Browser_test.go
index 0309f1f0b3..87fb9dfaa7 100644
--- a/plc4go/internal/cbus/Browser_test.go
+++ b/plc4go/internal/cbus/Browser_test.go
@@ -160,9 +160,7 @@ func TestBrowser_BrowseQuery(t *testing.T) {
assert.NoError(t, driver.Close())
})
connectionConnectResult := <-driver.GetConnection(transportUrl, map[string]transports.Transport{"test": transport}, map[string][]string{})
- if err := connectionConnectResult.GetErr(); err != nil {
- t.Fatal(err)
- }
+ require.NoError(t, connectionConnectResult.GetErr())
fields.connection = connectionConnectResult.GetConnection()
t.Cleanup(func() {
timer := time.NewTimer(10 * time.Second)
@@ -319,10 +317,7 @@ func TestBrowser_browseUnitInfo(t *testing.T) {
assert.NoError(t, driver.Close())
})
connectionConnectResult := <-driver.GetConnection(transportUrl, map[string]transports.Transport{"test": transport}, map[string][]string{})
- if err := connectionConnectResult.GetErr(); err != nil {
- t.Error(err)
- t.FailNow()
- }
+ require.NoError(t, connectionConnectResult.GetErr())
fields.connection = connectionConnectResult.GetConnection()
t.Cleanup(func() {
timer := time.NewTimer(10 * time.Second)
@@ -595,10 +590,7 @@ func TestBrowser_getInstalledUnitAddressBytes(t *testing.T) {
assert.NoError(t, driver.Close())
})
connectionConnectResult := <-driver.GetConnection(transportUrl, map[string]transports.Transport{"test": transport}, map[string][]string{})
- if err := connectionConnectResult.GetErr(); err != nil {
- t.Error(err)
- t.FailNow()
- }
+ require.NoError(t, connectionConnectResult.GetErr())
fields.connection = connectionConnectResult.GetConnection()
t.Cleanup(func() {
timer := time.NewTimer(6 * time.Second)