Repository: Tochemey/goakt
Pull request #440 (2024-08-25 20:58:04)

fix: fix wrong metric recording and cleanup code base

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/440?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 86.25%. Comparing base [(`7b070e1`)](https://app.codecov.io/gh/Tochemey/goakt/commit/7b070e15bcfa18e5f8c020479225ad838b5ce2ca?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) to head [(`60aae3c`)](https://app.codecov.io/gh/Tochemey/goakt/commit/60aae3c2eecc9c5ec7c479d7019f5c41813ebdf6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #440      +/-   ##
==========================================
+ Coverage   86.12%   86.25%   +0.13%     
==========================================
  Files          47       47              
  Lines        3560     3537      -23     
==========================================
- Hits         3066     3051      -15     
+ Misses        347      342       -5     
+ Partials      147      144       -3     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/440?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).


Patch:
diff --git a/actors/api.go b/actors/api.go
index 1ea5ae1..a07279f 100644
--- a/actors/api.go
+++ b/actors/api.go
@@ -48,30 +48,22 @@ func Ask(ctx context.Context, to *PID, message proto.Message, timeout time.Durat
 		return nil, ErrDead
 	}
 
-	var messageContext *ReceiveContext
-
-	switch msg := message.(type) {
-	case *internalpb.RemoteMessage:
-		var actual proto.Message
-		if actual, err = msg.GetMessage().UnmarshalNew(); err != nil {
-			return nil, ErrInvalidRemoteMessage(err)
-		}
-		messageContext = newReceiveContext(ctx, NoSender, to, actual).WithRemoteSender(msg.GetSender())
-	default:
-		messageContext = newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender)
+	receiveContext, err := toReceiveContext(ctx, to, message)
+	if err != nil {
+		return nil, err
 	}
 
-	to.doReceive(messageContext)
+	to.doReceive(receiveContext)
 
 	// await patiently to receive the response from the actor
 	select {
-	case response = <-messageContext.response:
-		to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
+	case response = <-receiveContext.response:
+		to.recordLatestReceiveDurationMetric(ctx)
 		return
 	case <-time.After(timeout):
-		to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
+		to.recordLatestReceiveDurationMetric(ctx)
 		err = ErrRequestTimeout
-		to.toDeadletterQueue(messageContext, err)
+		to.toDeadletterQueue(receiveContext, err)
 		return
 	}
 }
@@ -82,40 +74,26 @@ func Tell(ctx context.Context, to *PID, message proto.Message) error {
 		return ErrDead
 	}
 
-	var messageContext *ReceiveContext
-
-	switch msg := message.(type) {
-	case *internalpb.RemoteMessage:
-		var (
-			actual proto.Message
-			err    error
-		)
-
-		if actual, err = msg.GetMessage().UnmarshalNew(); err != nil {
-			return ErrInvalidRemoteMessage(err)
-		}
-		messageContext = newReceiveContext(ctx, NoSender, to, actual).WithRemoteSender(msg.GetSender())
-	default:
-		messageContext = newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender)
+	receiveContext, err := toReceiveContext(ctx, to, message)
+	if err != nil {
+		return err
 	}
 
-	to.doReceive(messageContext)
-	to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
+	to.doReceive(receiveContext)
+	to.recordLatestReceiveDurationMetric(ctx)
 	return nil
 }
 
 // BatchTell sends bulk asynchronous messages to an actor
+// The messages will be processed one after the other in the order they are sent
+// This is a design choice to follow the simple principle of one message at a time processing by actors.
 func BatchTell(ctx context.Context, to *PID, messages ...proto.Message) error {
-	if !to.IsRunning() {
-		return ErrDead
-	}
-
-	for i := 0; i < len(messages); i++ {
-		message := messages[i]
-		messageContext := newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender)
-		to.doReceive(messageContext)
+	// messages are processed one after the other
+	for _, mesage := range messages {
+		if err := Tell(ctx, to, mesage); err != nil {
+			return err
+		}
 	}
-	to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 	return nil
 }
 
@@ -123,25 +101,14 @@ func BatchTell(ctx context.Context, to *PID, messages ...proto.Message) error {
 // The messages will be processed one after the other in the order they are sent
 // This is a design choice to follow the simple principle of one message at a time processing by actors.
 func BatchAsk(ctx context.Context, to *PID, timeout time.Duration, messages ...proto.Message) (responses chan proto.Message, err error) {
-	if !to.IsRunning() {
-		return nil, ErrDead
-	}
-
 	responses = make(chan proto.Message, len(messages))
 	defer close(responses)
-
-	for i := 0; i < len(messages); i++ {
-		receiveContext := newReceiveContext(ctx, NoSender, to, messages[i])
-		to.doReceive(receiveContext)
-		select {
-		case result := <-receiveContext.response:
-			responses <- result
-			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
-		case <-time.After(timeout):
-			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
-			to.toDeadletterQueue(receiveContext, ErrRequestTimeout)
-			return nil, ErrRequestTimeout
+	for _, mesage := range messages {
+		response, err := Ask(ctx, to, mesage, timeout)
+		if err != nil {
+			return nil, err
 		}
+		responses <- response
 	}
 	return
 }
@@ -499,3 +466,17 @@ func RemoteSpawn(ctx context.Context, host string, port int, name, actorType str
 	}
 	return nil
 }
+
+// toReceiveContext creates a ReceiveContext provided a message and a receiver
+func toReceiveContext(ctx context.Context, to *PID, message proto.Message) (*ReceiveContext, error) {
+	switch msg := message.(type) {
+	case *internalpb.RemoteMessage:
+		actual, err := msg.GetMessage().UnmarshalNew()
+		if err != nil {
+			return nil, ErrInvalidRemoteMessage(err)
+		}
+		return newReceiveContext(ctx, NoSender, to, actual).WithRemoteSender(msg.GetSender()), nil
+	default:
+		return newReceiveContext(ctx, NoSender, to, message).WithRemoteSender(RemoteNoSender), nil
+	}
+}
diff --git a/actors/pid.go b/actors/pid.go
index 0b7a0dd..b04991e 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -29,6 +29,7 @@ import (
 	"errors"
 	"fmt"
 	stdhttp "net/http"
+	"os"
 	"strings"
 	"sync"
 	"time"
@@ -81,8 +82,9 @@ type PID struct {
 
 	// helps determine whether the actor should handle public or not.
 	running atomic.Bool
-	// is captured whenever a mail is sent to the actor
-	lastProcessingTime atomic.Time
+
+	latestReceiveTime     atomic.Time
+	latestReceiveDuration atomic.Duration
 
 	// specifies at what point in time to passivate the actor.
 	// when the actor is passivated it is stopped which means it does not consume
@@ -125,9 +127,6 @@ type PID struct {
 	// specifies the logger to use
 	logger log.Logger
 
-	// specifies the last processing message duration
-	lastReceivedDuration atomic.Duration
-
 	// fieldsLocker that helps synchronize the pid in a concurrent environment
 	// this helps protect the pid fields accessibility
 	fieldsLocker *sync.RWMutex
@@ -160,6 +159,9 @@ type PID struct {
 	processedCount *atomic.Int64
 	metricEnabled  atomic.Bool
 	metrics        *metric.ActorMetric
+
+	watcherNotificationChan        chan error
+	watchersNotificationStopSignal chan types.Unit
 }
 
 // newPID creates a new pid
@@ -175,32 +177,34 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 	}
 
 	p := &PID{
-		actor:                actor,
-		lastProcessingTime:   atomic.Time{},
-		haltPassivationLnr:   make(chan types.Unit, 1),
-		logger:               log.DefaultLogger,
-		children:             newPIDMap(10),
-		supervisorDirective:  DefaultSupervisoryStrategy,
-		watchersList:         slice.New[*watcher](),
-		watchedList:          newPIDMap(10),
-		telemetry:            telemetry.New(),
-		actorPath:            actorPath,
-		fieldsLocker:         &sync.RWMutex{},
-		stopLocker:           &sync.Mutex{},
-		httpClient:           http.NewClient(),
-		mailbox:              newMailbox(),
-		stashBuffer:          nil,
-		stashLocker:          &sync.Mutex{},
-		eventsStream:         nil,
-		restartCount:         atomic.NewInt64(0),
-		childrenCount:        atomic.NewInt64(0),
-		processedCount:       atomic.NewInt64(0),
-		processingTimeLocker: new(sync.Mutex),
+		actor:                          actor,
+		latestReceiveTime:              atomic.Time{},
+		haltPassivationLnr:             make(chan types.Unit, 1),
+		logger:                         log.New(log.InfoLevel, os.Stderr),
+		children:                       newPIDMap(10),
+		supervisorDirective:            DefaultSupervisoryStrategy,
+		watchersList:                   slice.New[*watcher](),
+		watchedList:                    newPIDMap(10),
+		telemetry:                      telemetry.New(),
+		actorPath:                      actorPath,
+		fieldsLocker:                   new(sync.RWMutex),
+		stopLocker:                     new(sync.Mutex),
+		httpClient:                     http.NewClient(),
+		mailbox:                        newMailbox(),
+		stashBuffer:                    nil,
+		stashLocker:                    &sync.Mutex{},
+		eventsStream:                   nil,
+		restartCount:                   atomic.NewInt64(0),
+		childrenCount:                  atomic.NewInt64(0),
+		processedCount:                 atomic.NewInt64(0),
+		processingTimeLocker:           new(sync.Mutex),
+		watcherNotificationChan:        make(chan error, 1),
+		watchersNotificationStopSignal: make(chan types.Unit, 1),
 	}
 
 	p.initMaxRetries.Store(DefaultInitMaxRetries)
 	p.shutdownTimeout.Store(DefaultShutdownTimeout)
-	p.lastReceivedDuration.Store(0)
+	p.latestReceiveDuration.Store(0)
 	p.running.Store(false)
 	p.passivateAfter.Store(DefaultPassivationTimeout)
 	p.askTimeout.Store(DefaultAskTimeout)
@@ -220,7 +224,7 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 	}
 
 	go p.receive()
-
+	go p.notifyWatchers()
 	if p.passivateAfter.Load() > 0 {
 		go p.passivationLoop()
 	}
@@ -394,6 +398,7 @@ func (x *PID) Restart(ctx context.Context) error {
 		return err
 	}
 	go x.receive()
+	go x.notifyWatchers()
 	if x.passivateAfter.Load() > 0 {
 		go x.passivationLoop()
 	}
@@ -522,10 +527,10 @@ func (x *PID) Ask(ctx context.Context, to *PID, message proto.Message) (response
 
 	select {
 	case result := <-receiveContext.response:
-		x.recordLastReceivedDurationMetric(ctx)
+		x.recordLatestReceiveDurationMetric(ctx)
 		return result, nil
 	case <-time.After(timeout):
-		x.recordLastReceivedDurationMetric(ctx)
+		x.recordLatestReceiveDurationMetric(ctx)
 		err = ErrRequestTimeout
 		x.toDeadletterQueue(receiveContext, err)
 		return nil, err
@@ -539,7 +544,7 @@ func (x *PID) Tell(ctx context.Context, to *PID, message proto.Message) error {
 	}
 
 	to.doReceive(newReceiveContext(ctx, x, to, message))
-	x.recordLastReceivedDurationMetric(ctx)
+	x.recordLatestReceiveDurationMetric(ctx)
 	return nil
 }
 
@@ -548,19 +553,11 @@ func (x *PID) Tell(ctx context.Context, to *PID, message proto.Message) error {
 // This is a design choice to follow the simple principle of one message at a time processing by actors.
 // When BatchTell encounter a single message it will fall back to a Tell call.
 func (x *PID) BatchTell(ctx context.Context, to *PID, messages ...proto.Message) error {
-	if !to.IsRunning() {
-		return ErrDead
-	}
-
-	if len(messages) == 1 {
-		return x.Tell(ctx, to, messages[0])
-	}
-
-	for i := 0; i < len(messages); i++ {
-		to.doReceive(newReceiveContext(ctx, x, to, messages[i]))
-		x.recordLastReceivedDurationMetric(ctx)
+	for _, message := range messages {
+		if err := x.Tell(ctx, to, message); err != nil {
+			return err
+		}
 	}
-
 	return nil
 }
 
@@ -568,10 +565,6 @@ func (x *PID) BatchTell(ctx context.Context, to *PID, messages ...proto.Message)
 // The messages will be processed one after the other in the order they are sent.
 // This is a design choice to follow the simple principle of one message at a time processing by actors.
 func (x *PID) BatchAsk(ctx context.Context, to *PID, messages ...proto.Message) (responses chan proto.Message, err error) {
-	if !to.IsRunning() {
-		return nil, ErrDead
-	}
-
 	responses = make(chan proto.Message, len(messages))
 	defer close(responses)
 
@@ -1011,7 +1004,7 @@ func (x *PID) watchees() *pidMap {
 
 // doReceive pushes a given message to the actor receiveContextBuffer
 func (x *PID) doReceive(receiveCtx *ReceiveContext) {
-	x.lastProcessingTime.Store(time.Now())
+	x.latestReceiveTime.Store(time.Now())
 	x.mailbox.Push(receiveCtx)
 }
 
@@ -1049,12 +1042,12 @@ func (x *PID) init(ctx context.Context) error {
 
 // reset re-initializes the actor PID
 func (x *PID) reset() {
-	x.lastProcessingTime.Store(time.Time{})
+	x.latestReceiveTime.Store(time.Time{})
 	x.passivateAfter.Store(DefaultPassivationTimeout)
 	x.askTimeout.Store(DefaultAskTimeout)
 	x.shutdownTimeout.Store(DefaultShutdownTimeout)
 	x.initMaxRetries.Store(DefaultInitMaxRetries)
-	x.lastReceivedDuration.Store(0)
+	x.latestReceiveDuration.Store(0)
 	x.initTimeout.Store(DefaultInitTimeout)
 	x.children.reset()
 	x.watchersList.Reset()
@@ -1125,7 +1118,6 @@ func (x *PID) receive() {
 	for x.running.Load() {
 		received := x.mailbox.Pop()
 		if received == nil {
-			//runtime.Gosched()
 			continue
 		}
 
@@ -1141,14 +1133,10 @@ func (x *PID) receive() {
 
 // handleReceived picks the right behavior and processes the message
 func (x *PID) handleReceived(received *ReceiveContext) {
-	// handle panic or any error during processing
 	defer x.recovery(received)
-	// pick the current behvior from the stack
-	// send the message to the current actor behavior
+
 	if behavior := x.behaviorStack.Peek(); behavior != nil {
-		// set the total messages processed
 		x.processedCount.Inc()
-		// call the behavior to handle the message
 		behavior(received)
 	}
 }
@@ -1156,11 +1144,11 @@ func (x *PID) handleReceived(received *ReceiveContext) {
 // recovery is called upon after message is processed
 func (x *PID) recovery(received *ReceiveContext) {
 	if r := recover(); r != nil {
-		x.notifyWatchers(fmt.Errorf("%s", r))
+		x.watcherNotificationChan <- fmt.Errorf("%s", r)
 		return
 	}
-	// no panic
-	x.notifyWatchers(received.getError())
+	// no panic or recommended way to handle error
+	x.watcherNotificationChan <- received.getError()
 }
 
 // passivationLoop checks whether the actor is processing public or not.
@@ -1176,7 +1164,7 @@ func (x *PID) passivationLoop() {
 		for {
 			select {
 			case <-ticker.C:
-				idleTime := time.Since(x.lastProcessingTime.Load())
+				idleTime := time.Since(x.latestReceiveTime.Load())
 				if idleTime >= x.passivateAfter.Load() {
 					tickerStopSig <- types.Unit{}
 					return
@@ -1282,7 +1270,7 @@ func (x *PID) doStop(ctx context.Context) error {
 
 	<-tickerStopSig
 	x.httpClient.CloseIdleConnections()
-
+	x.watchersNotificationStopSignal <- types.Unit{}
 	x.freeWatchees(ctx)
 	x.freeChildren(ctx)
 	x.freeWatchers(ctx)
@@ -1312,10 +1300,17 @@ func (x *PID) removeChild(pid *PID) {
 }
 
 // notifyWatchers send error to watchers
-func (x *PID) notifyWatchers(err error) {
-	if err != nil {
-		for _, item := range x.watchers().Items() {
-			item.Value.ErrChan <- err
+func (x *PID) notifyWatchers() {
+	for {
+		select {
+		case err := <-x.watcherNotificationChan:
+			if err != nil {
+				for _, item := range x.watchers().Items() {
+					item.Value.ErrChan <- err
+				}
+			}
+		case <-x.watchersNotificationStopSignal:
+			return
 		}
 	}
 }
@@ -1402,21 +1397,6 @@ func (x *PID) remotingClient(host string, port int) (internalpbconnect.RemotingS
 	), nil
 }
 
-// getLastProcessingTime returns the last processing time
-func (x *PID) getLastProcessingTime() time.Time {
-	x.processingTimeLocker.Lock()
-	processingTime := x.lastProcessingTime.Load()
-	x.processingTimeLocker.Unlock()
-	return processingTime
-}
-
-// setLastProcessingDuration sets the last processing duration
-func (x *PID) setLastProcessingDuration(d time.Duration) {
-	x.processingTimeLocker.Lock()
-	x.lastReceivedDuration.Store(d)
-	x.processingTimeLocker.Unlock()
-}
-
 // handleCompletion processes a long-running task and pipe the result to
 // the completion receiver
 func (x *PID) handleCompletion(ctx context.Context, completion *taskCompletion) {
@@ -1519,9 +1499,10 @@ func (x *PID) handleRestartDirective(cid *PID, maxRetries uint32, timeout time.D
 	x.Watch(cid)
 }
 
-func (x *PID) recordLastReceivedDurationMetric(ctx context.Context) {
-	x.lastReceivedDuration.Store(time.Since(x.lastProcessingTime.Load()))
+func (x *PID) recordLatestReceiveDurationMetric(ctx context.Context) {
+	// record processing time
+	x.latestReceiveDuration.Store(time.Since(x.latestReceiveTime.Load()))
 	if x.metricEnabled.Load() {
-		x.metrics.LastReceivedDuration().Record(ctx, x.lastReceivedDuration.Load().Milliseconds())
+		x.metrics.LastReceivedDuration().Record(ctx, x.latestReceiveDuration.Load().Milliseconds())
 	}
 }
diff --git a/bench/benchmark.go b/bench/benchmark.go
index 2438dbc..a3095de 100644
--- a/bench/benchmark.go
+++ b/bench/benchmark.go
@@ -27,7 +27,6 @@ package bench
 import (
 	"context"
 	"fmt"
-	"math/rand/v2"
 	"sync"
 	"time"
 
@@ -46,6 +45,11 @@ var (
 	totalRecv *atomic.Int64
 )
 
+func init() {
+	totalSent = atomic.NewInt64(0)
+	totalRecv = atomic.NewInt64(0)
+}
+
 type Benchmarker struct {
 }
 
@@ -79,17 +83,15 @@ type Benchmark struct {
 	workersCount int
 	// duration specifies how long the load testing will run
 	duration time.Duration
-	pids     []*actors.PID
+	pid      *actors.PID
 	system   actors.ActorSystem
 }
 
 // NewBenchmark creates an instance of Loader
-func NewBenchmark(actorsCount, workersCount int, duration time.Duration) *Benchmark {
+func NewBenchmark(workersCount int, duration time.Duration) *Benchmark {
 	return &Benchmark{
-		actorsCount:  actorsCount,
 		workersCount: workersCount,
 		duration:     duration,
-		pids:         make([]*actors.PID, 0, actorsCount),
 	}
 }
 
@@ -110,14 +112,13 @@ func (b *Benchmark) Start(ctx context.Context) error {
 	// wait for the actor system to properly start
 	time.Sleep(time.Second)
 
-	for i := 0; i < b.actorsCount; i++ {
-		actorName := fmt.Sprintf("actor-%d", i)
-		pid, err := b.system.Spawn(ctx, actorName, &Benchmarker{})
-		if err != nil {
-			return err
-		}
-		b.pids = append(b.pids, pid)
+	pid, err := b.system.Spawn(ctx, "Benchmarker", &Benchmarker{})
+	if err != nil {
+		return err
 	}
+
+	b.pid = pid
+
 	// wait for the actors to properly start
 	time.Sleep(time.Second)
 	return nil
@@ -138,10 +139,7 @@ func (b *Benchmark) BenchTell(ctx context.Context) error {
 		go func() {
 			defer wg.Done()
 			for time.Now().Before(deadline) {
-				// randomly pick and actor
-				pid := b.pids[rand.IntN(len(b.pids))] //nolint:gosec
-				// send a message
-				_ = actors.Tell(ctx, pid, new(benchmarkpb.BenchTell))
+				_ = actors.Tell(ctx, b.pid, new(benchmarkpb.BenchTell))
 				// increase sent counter
 				totalSent.Inc()
 			}
@@ -166,10 +164,7 @@ func (b *Benchmark) BenchAsk(ctx context.Context) error {
 		go func() {
 			defer wg.Done()
 			for time.Now().Before(deadline) {
-				// randomly pick and actor
-				pid := b.pids[rand.IntN(len(b.pids))] //nolint:gosec
-				// send a message
-				_, _ = actors.Ask(ctx, pid, new(benchmarkpb.BenchRequest), receivingTimeout)
+				_, _ = actors.Ask(ctx, b.pid, new(benchmarkpb.BenchRequest), receivingTimeout)
 				// increase sent counter
 				totalSent.Add(1)
 			}


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index cb5d649..381f279 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -1013,8 +1013,8 @@ func TestActorSystem(t *testing.T) {
 		got := &metricdata.ResourceMetrics{}
 		err = r.Collect(ctx, got)
 		require.NoError(t, err)
-		assert.Len(t, got.ScopeMetrics, 1)
-		assert.Len(t, got.ScopeMetrics[0].Metrics, 5)
+		require.Len(t, got.ScopeMetrics, 1)
+		require.Len(t, got.ScopeMetrics[0].Metrics, 6)
 
 		expected := []string{
 			"actor_child_count",
@@ -1022,6 +1022,7 @@ func TestActorSystem(t *testing.T) {
 			"actor_restart_count",
 			"actors_count",
 			"actor_processed_count",
+			"actor_received_duration",
 		}
 		// sort the array
 		sort.Strings(expected)
diff --git a/bench/benchmark_test.go b/bench/benchmark_test.go
index 9519c70..c85de96 100644
--- a/bench/benchmark_test.go
+++ b/bench/benchmark_test.go
@@ -63,18 +63,24 @@ func BenchmarkActor(b *testing.B) {
 		// wait for actors to start properly
 		time.Sleep(1 * time.Second)
 
-		runParallel(b, func(pb *testing.PB) {
+		b.SetParallelism(7)
+		b.ResetTimer()
+		b.ReportAllocs()
+		start := time.Now()
+		b.RunParallel(func(pb *testing.PB) {
 			for pb.Next() {
-				// send a message to the actor
 				_ = actors.Tell(ctx, pid, new(benchmarkpb.BenchTell))
 			}
 		})
 
+		opsPerSec := float64(b.N) / time.Since(start).Seconds()
+		b.ReportMetric(opsPerSec, "ops/s")
+
 		_ = pid.Shutdown(ctx)
 		_ = actorSystem.Stop(ctx)
 	})
+
 	b.Run("ask", func(b *testing.B) {
-		b.Skip("")
 		ctx := context.TODO()
 		// create the actor system
 		actorSystem, _ := actors.NewActorSystem("bench",
@@ -98,12 +104,18 @@ func BenchmarkActor(b *testing.B) {
 		// wait for actors to start properly
 		time.Sleep(1 * time.Second)
 
-		runParallel(b, func(pb *testing.PB) {
+		b.SetParallelism(7)
+		b.ResetTimer()
+		b.ReportAllocs()
+		start := time.Now()
+
+		b.RunParallel(func(pb *testing.PB) {
 			for pb.Next() {
-				// send a message to the actor
 				_, _ = actors.Ask(ctx, pid, new(benchmarkpb.BenchRequest), receivingTimeout)
 			}
 		})
+		opsPerSec := float64(b.N) / time.Since(start).Seconds()
+		b.ReportMetric(opsPerSec, "ops/s")
 
 		_ = pid.Shutdown(ctx)
 		_ = actorSystem.Stop(ctx)
@@ -111,18 +123,16 @@ func BenchmarkActor(b *testing.B) {
 }
 
 func TestBenchmark_BenchTell(t *testing.T) {
-	// remove this line for local benchmark
-	t.Skip("only run this test for local benchmark")
+	t.Skip("")
 	ctx := context.TODO()
 
 	totalSent = atomic.NewInt64(0)
 	totalRecv = atomic.NewInt64(0)
 
-	actorsCount := 2000
-	workersCount := 20
+	workersCount := 100
 	duration := 30 * time.Second
 
-	benchmark := NewBenchmark(actorsCount, workersCount, duration)
+	benchmark := NewBenchmark(workersCount, duration)
 	if err := benchmark.Start(ctx); err != nil {
 		t.Fatal(err)
 	}
@@ -132,7 +142,6 @@ func TestBenchmark_BenchTell(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	fmt.Printf("Total actors spawned: (%d)\n", actorsCount)
 	fmt.Printf("Total workers: (%d), total messages sent: (%d), total messages received: (%d) - duration: (%v)\n", workersCount, totalSent.Load(), totalRecv.Load(), duration)
 	fmt.Printf("Messages per second: (%d)\n", totalRecv.Load()/int64(duration.Seconds()))
 	t.Cleanup(func() {
@@ -143,18 +152,15 @@ func TestBenchmark_BenchTell(t *testing.T) {
 }
 
 func TestBenchmark_BenchAsk(t *testing.T) {
-	// remove this line for local benchmark
-	t.Skip("only run this test for local benchmark")
-
 	ctx := context.TODO()
-
+	t.Skip("")
 	totalSent = atomic.NewInt64(0)
 	totalRecv = atomic.NewInt64(0)
-	actorsCount := 2000
-	workersCount := 20
+
+	workersCount := 100
 	duration := 30 * time.Second
 
-	benchmark := NewBenchmark(actorsCount, workersCount, duration)
+	benchmark := NewBenchmark(workersCount, duration)
 	if err := benchmark.Start(ctx); err != nil {
 		t.Fatal(err)
 	}
@@ -164,7 +170,6 @@ func TestBenchmark_BenchAsk(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	fmt.Printf("Total actors spawned: (%d)\n", actorsCount)
 	fmt.Printf("Total workers: (%d), total messages sent: (%d), total messages received: (%d) - duration: (%v)\n", workersCount, totalSent.Load(), totalRecv.Load(), duration)
 	fmt.Printf("Messages per second: (%d)\n", totalRecv.Load()/int64(duration.Seconds()))
 	t.Cleanup(func() {
@@ -173,12 +178,3 @@ func TestBenchmark_BenchAsk(t *testing.T) {
 		}
 	})
 }
-
-func runParallel(b *testing.B, benchFn func(pb *testing.PB)) {
-	b.ReportAllocs()
-	b.ResetTimer()
-	start := time.Now()
-	b.RunParallel(benchFn)
-	opsPerSec := float64(b.N) / time.Since(start).Seconds()
-	b.ReportMetric(opsPerSec, "ops/s")
-}