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:38 UTC
[plc4x] 01/02: refactor(plc4go/spi): avoid issues when using executor and logging
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")