Repository: Tochemey/goakt
Pull request #358 (2024-06-09 12:22:43)

perf: increase default mailbox size and stash size.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/358?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Attention: Patch coverage is `99.02913%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 85.01%. Comparing base [(`23db4ad`)](https://app.codecov.io/gh/Tochemey/goakt/commit/23db4ad69d77fd0c5f32f85bb987769c6c0c13eb?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) to head [(`0ef3f07`)](https://app.codecov.io/gh/Tochemey/goakt/commit/0ef3f074b157591729803ef606d1750e4e88bdf3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).

| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/358?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Patch % | Lines |
|---|---|---|
| [actors/pid.go](https://app.codecov.io/gh/Tochemey/goakt/pull/358?src=pr&el=tree&filepath=actors%2Fpid.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL3BpZC5nbw==) | 98.55% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/358?src=pr&el=tree&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     #358      +/-   ##
==========================================
+ Coverage   84.96%   85.01%   +0.04%     
==========================================
  Files          36       36              
  Lines        3220     3237      +17     
==========================================
+ Hits         2736     2752      +16     
  Misses        366      366              
- Partials      118      119       +1     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/358?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/actor_system.go b/actors/actor_system.go
index dbdbf75..7a41032 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -185,7 +185,7 @@ type actorSystem struct {
 	partitionHasher    hash.Hasher
 
 	// help protect some the fields to set
-	mutex sync.Mutex
+	locker sync.Mutex
 	// specifies actors mailbox size
 	mailboxSize uint64
 	// specifies the mailbox to use for the actors
@@ -239,9 +239,9 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		actorInitMaxRetries: DefaultInitMaxRetries,
 		supervisorStrategy:  DefaultSupervisoryStrategy,
 		telemetry:           telemetry.New(),
-		mutex:               sync.Mutex{},
+		locker:              sync.Mutex{},
 		shutdownTimeout:     DefaultShutdownTimeout,
-		mailboxSize:         defaultMailboxSize,
+		mailboxSize:         DefaultMailboxSize,
 		housekeeperStopSig:  make(chan types.Unit, 1),
 		eventsStream:        eventstream.New(),
 		partitionHasher:     hash.DefaultHasher(),
@@ -294,8 +294,8 @@ func (x *actorSystem) Deregister(ctx context.Context, actor Actor) error {
 	_, span := x.tracer.Start(ctx, "Deregister")
 	defer span.End()
 
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 
 	if !x.started.Load() {
 		return ErrActorSystemNotStarted
@@ -310,8 +310,8 @@ func (x *actorSystem) Register(ctx context.Context, actor Actor) error {
 	_, span := x.tracer.Start(ctx, "Register")
 	defer span.End()
 
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 
 	if !x.started.Load() {
 		return ErrActorSystemNotStarted
@@ -612,23 +612,23 @@ func (x *actorSystem) ReSpawn(ctx context.Context, name string) (PID, error) {
 
 // Name returns the actor system name
 func (x *actorSystem) Name() string {
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 	return x.name
 }
 
 // Actors returns the list of Actors that are alive in the actor system
 func (x *actorSystem) Actors() []PID {
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 	return x.actors.pids()
 }
 
 // PeerAddress returns the actor system address known in the cluster. That address is used by other nodes to communicate with the actor system.
 // This address is empty when cluster mode is not activated
 func (x *actorSystem) PeerAddress() string {
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 	if x.clusterEnabled.Load() {
 		return x.cluster.AdvertisedAddress()
 	}
@@ -643,8 +643,8 @@ func (x *actorSystem) ActorOf(ctx context.Context, actorName string) (addr *goak
 	spanCtx, span := x.tracer.Start(ctx, "ActorOf")
 	defer span.End()
 
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 
 	if !x.started.Load() {
 		span.SetStatus(codes.Error, "ActorOf")
@@ -697,8 +697,8 @@ func (x *actorSystem) ActorOf(ctx context.Context, actorName string) (addr *goak
 // LocalActor returns the reference of a local actor.
 // A local actor is an actor that reside on the same node where the given actor system is running
 func (x *actorSystem) LocalActor(actorName string) (PID, error) {
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 
 	if !x.started.Load() {
 		return nil, ErrActorSystemNotStarted
@@ -722,8 +722,8 @@ func (x *actorSystem) RemoteActor(ctx context.Context, actorName string) (addr *
 	spanCtx, span := x.tracer.Start(ctx, "RemoteActor")
 	defer span.End()
 
-	x.mutex.Lock()
-	defer x.mutex.Unlock()
+	x.locker.Lock()
+	defer x.locker.Unlock()
 
 	if !x.started.Load() {
 		e := ErrActorSystemNotStarted
@@ -1128,7 +1128,7 @@ func (x *actorSystem) enableClustering(ctx context.Context) error {
 
 	x.logger.Info("cluster engine successfully started...")
 
-	x.mutex.Lock()
+	x.locker.Lock()
 	x.cluster = clusterEngine
 	x.clusterEventsChan = clusterEngine.Events()
 	x.redistributionChan = make(chan *cluster.Event, 1)
@@ -1136,7 +1136,7 @@ func (x *actorSystem) enableClustering(ctx context.Context) error {
 		x.registry.Register(kind)
 		x.logger.Infof("cluster kind=(%s) registered", types.NameOf(kind))
 	}
-	x.mutex.Unlock()
+	x.locker.Unlock()
 
 	go x.clusterEventsLoop()
 	go x.clusterReplicationLoop()
@@ -1216,9 +1216,9 @@ func (x *actorSystem) enableRemoting(ctx context.Context) {
 		}
 	}()
 
-	x.mutex.Lock()
+	x.locker.Lock()
 	x.remotingServer = server
-	x.mutex.Unlock()
+	x.locker.Unlock()
 
 	x.logger.Info("remoting enabled...:)")
 }
diff --git a/actors/api.go b/actors/api.go
index 66e6eb7..31f50ef 100644
--- a/actors/api.go
+++ b/actors/api.go
@@ -66,8 +66,10 @@ func Ask(ctx context.Context, to PID, message proto.Message, timeout time.Durati
 	for await := time.After(timeout); ; {
 		select {
 		case response = <-messageContext.response:
+			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 			return
 		case <-await:
+			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 			err = ErrRequestTimeout
 			to.handleError(messageContext, err)
 			return
@@ -99,7 +101,7 @@ func Tell(ctx context.Context, to PID, message proto.Message) error {
 	}
 
 	to.doReceive(messageContext)
-
+	to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 	return nil
 }
 
@@ -114,6 +116,7 @@ func BatchTell(ctx context.Context, to PID, messages ...proto.Message) error {
 		messageContext := newReceiveContext(ctx, NoSender, to, message, true).WithRemoteSender(RemoteNoSender)
 		to.doReceive(messageContext)
 	}
+	to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 	return nil
 }
 
@@ -138,9 +141,11 @@ func BatchAsk(ctx context.Context, to PID, timeout time.Duration, messages ...pr
 		for await := time.After(timeout); ; {
 			select {
 			case resp := <-messageContext.response:
+				to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 				responses <- resp
 				break timerLoop
 			case <-await:
+				to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 				err = ErrRequestTimeout
 				to.handleError(messageContext, err)
 				return
diff --git a/actors/constants.go b/actors/constants.go
index 6c8bdd2..df26862 100644
--- a/actors/constants.go
+++ b/actors/constants.go
@@ -53,9 +53,11 @@ const (
 	DefaultShutdownTimeout = 2 * time.Second
 	// DefaultInitTimeout defines the default init timeout
 	DefaultInitTimeout = time.Second
-
-	defaultMailboxSize = 4096
-	eventsTopic        = "topic.events"
+	// DefaultMailboxSize defines the default mailbox size
+	DefaultMailboxSize = 1_000_000
+	// DefaultStashCapacity defines the default stash size
+	DefaultStashCapacity = 1_000
+	eventsTopic          = "topic.events"
 )
 
 // NoSender means that there is no sender
diff --git a/actors/pid.go b/actors/pid.go
index acaf1e0..5dc520e 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -165,6 +165,9 @@ type PID interface {
 	handleError(receiveCtx ReceiveContext, err error)
 	// removeChild is a utility function to remove child actor
 	removeChild(pid PID)
+
+	getLastProcessingTime() time.Time
+	setLastProcessingDuration(d time.Duration)
 }
 
 // pid specifies an actor unique process
@@ -225,10 +228,12 @@ type pid struct {
 	// specifies the last processing message duration
 	lastProcessingDuration atomic.Duration
 
-	// rwMutex that helps synchronize the pid in a concurrent environment
+	// rwLocker that helps synchronize the pid in a concurrent environment
 	// this helps protect the pid fields accessibility
-	rwMutex   *sync.RWMutex
-	stopMutex *sync.Mutex
+	rwLocker *sync.RWMutex
+
+	stopLocker           *sync.Mutex
+	processingTimeLocker *sync.Mutex
 
 	// supervisor strategy
 	supervisorStrategy StrategyDirective
@@ -245,7 +250,7 @@ type pid struct {
 	// stash settings
 	stashBuffer   Mailbox
 	stashCapacity atomic.Uint64
-	stashMutex    *sync.Mutex
+	stashLocker   *sync.Mutex
 
 	// define an events stream
 	eventsStream *eventstream.EventsStream
@@ -266,33 +271,34 @@ var _ PID = (*pid)(nil)
 // newPID creates a new pid
 func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption) (*pid, error) {
 	p := &pid{
-		Actor:              actor,
-		lastProcessingTime: atomic.Time{},
-		shutdownSignal:     make(chan types.Unit, 1),
-		haltPassivationLnr: make(chan types.Unit, 1),
-		logger:             log.DefaultLogger,
-		mailboxSize:        defaultMailboxSize,
-		children:           newPIDMap(10),
-		supervisorStrategy: DefaultSupervisoryStrategy,
-		watchersList:       slices.NewConcurrentSlice[*watcher](),
-		telemetry:          telemetry.New(),
-		actorPath:          actorPath,
-		rwMutex:            &sync.RWMutex{},
-		stopMutex:          &sync.Mutex{},
-		httpClient:         http.NewClient(),
-		mailbox:            nil,
-		stashBuffer:        nil,
-		stashMutex:         &sync.Mutex{},
-		eventsStream:       nil,
-		tracer:             noop.NewTracerProvider().Tracer("PID"),
-		restartCount:       atomic.NewInt64(0),
-		childrenCount:      atomic.NewInt64(0),
+		Actor:                actor,
+		lastProcessingTime:   atomic.Time{},
+		shutdownSignal:       make(chan types.Unit, 1),
+		haltPassivationLnr:   make(chan types.Unit, 1),
+		logger:               log.DefaultLogger,
+		mailboxSize:          DefaultMailboxSize,
+		children:             newPIDMap(10),
+		supervisorStrategy:   DefaultSupervisoryStrategy,
+		watchersList:         slices.NewConcurrentSlice[*watcher](),
+		telemetry:            telemetry.New(),
+		actorPath:            actorPath,
+		rwLocker:             &sync.RWMutex{},
+		stopLocker:           &sync.Mutex{},
+		httpClient:           http.NewClient(),
+		mailbox:              nil,
+		stashBuffer:          nil,
+		stashLocker:          &sync.Mutex{},
+		eventsStream:         nil,
+		tracer:               noop.NewTracerProvider().Tracer("PID"),
+		restartCount:         atomic.NewInt64(0),
+		childrenCount:        atomic.NewInt64(0),
+		processingTimeLocker: new(sync.Mutex),
 	}
 
 	p.initMaxRetries.Store(DefaultInitMaxRetries)
 	p.shutdownTimeout.Store(DefaultShutdownTimeout)
 	p.lastProcessingDuration.Store(0)
-	p.stashCapacity.Store(0)
+	p.stashCapacity.Store(DefaultStashCapacity)
 	p.isRunning.Store(false)
 	p.passivateAfter.Store(DefaultPassivationTimeout)
 	p.replyTimeout.Store(DefaultReplyTimeout)
@@ -377,9 +383,9 @@ func (p *pid) Child(name string) (PID, error) {
 
 // Children returns the list of all the children of the given actor that are still alive or an empty list
 func (p *pid) Children() []PID {
-	p.rwMutex.RLock()
+	p.rwLocker.RLock()
 	children := p.children.pids()
-	p.rwMutex.RUnlock()
+	p.rwLocker.RUnlock()
 
 	cids := make([]PID, 0, len(children))
 	for _, child := range children {
@@ -409,9 +415,9 @@ func (p *pid) Stop(ctx context.Context, cid PID) error {
 		return ErrUndefinedActor
 	}
 
-	p.rwMutex.RLock()
+	p.rwLocker.RLock()
 	children := p.children
-	p.rwMutex.RUnlock()
+	p.rwLocker.RUnlock()
 
 	if cid, ok := children.get(cid.ActorPath()); ok {
 		desc := fmt.Sprintf("child.[%s] Shutdown", cid.ActorPath())
@@ -439,17 +445,17 @@ func (p *pid) IsRunning() bool {
 
 // ActorSystem returns the actor system
 func (p *pid) ActorSystem() ActorSystem {
-	p.rwMutex.RLock()
+	p.rwLocker.RLock()
 	sys := p.system
-	p.rwMutex.RUnlock()
+	p.rwLocker.RUnlock()
 	return sys
 }
 
 // ActorPath returns the path of the actor
 func (p *pid) ActorPath() *Path {
-	p.rwMutex.RLock()
+	p.rwLocker.RLock()
 	path := p.actorPath
-	p.rwMutex.RUnlock()
+	p.rwLocker.RUnlock()
 	return path
 }
 
@@ -522,16 +528,16 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 
 	childActorPath := NewPath(name, p.ActorPath().Address()).WithParent(p.ActorPath())
 
-	p.rwMutex.RLock()
+	p.rwLocker.RLock()
 	children := p.children
-	p.rwMutex.RUnlock()
+	p.rwLocker.RUnlock()
 
 	if cid, ok := children.get(childActorPath); ok {
 		return cid, nil
 	}
 
-	p.rwMutex.Lock()
-	defer p.rwMutex.Unlock()
+	p.rwLocker.Lock()
+	defer p.rwLocker.Unlock()
 
 	// create the child actor options child inherit parent's options
 	opts := []pidOption{
@@ -611,9 +617,11 @@ func (p *pid) Ask(ctx context.Context, to PID, message proto.Message) (response
 	for await := time.After(p.replyTimeout.Load()); ; {
 		select {
 		case response = <-messageContext.response:
+			p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
 			span.SetStatus(codes.Ok, "Ask")
 			return
 		case <-await:
+			p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
 			err = ErrRequestTimeout
 			span.SetStatus(codes.Error, "Ask")
 			span.RecordError(err)
@@ -636,7 +644,7 @@ func (p *pid) Tell(ctx context.Context, to PID, message proto.Message) error {
 
 	messageContext := newReceiveContext(spanCtx, p, to, message, true)
 	to.doReceive(messageContext)
-
+	p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
 	span.SetStatus(codes.Ok, "Tell")
 	return nil
 }
@@ -666,6 +674,7 @@ func (p *pid) BatchTell(ctx context.Context, to PID, messages ...proto.Message)
 		to.doReceive(messageContext)
 	}
 
+	p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
 	span.SetStatus(codes.Ok, "BatchTell")
 	return nil
 }
@@ -696,10 +705,12 @@ func (p *pid) BatchAsk(ctx context.Context, to PID, messages ...proto.Message) (
 		for await := time.After(p.replyTimeout.Load()); ; {
 			select {
 			case resp := <-messageContext.response:
+				p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
 				responses <- resp
 				span.SetStatus(codes.Ok, "BatchAsk")
 				break timerLoop
 			case <-await:
+				p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
 				err = ErrRequestTimeout
 				span.SetStatus(codes.Error, "BatchAsk")
 				span.RecordError(err)
@@ -1070,8 +1081,8 @@ func (p *pid) Shutdown(ctx context.Context) error {
 	spanCtx, span := p.tracer.Start(ctx, "Shutdown")
 	defer span.End()
 
-	p.stopMutex.Lock()
-	defer p.stopMutex.Unlock()
+	p.stopLocker.Lock()
+	defer p.stopLocker.Unlock()
 
 	p.logger.Info("Shutdown process has started...")
 
@@ -1133,21 +1144,14 @@ func (p *pid) watchers() *slices.ConcurrentSlice[*watcher] {
 
 // doReceive pushes a given message to the actor receiveContextBuffer
 func (p *pid) doReceive(ctx ReceiveContext) {
-	p.rwMutex.Lock()
-	defer p.rwMutex.Unlock()
+	p.rwLocker.Lock()
+	mailbox := p.mailbox
+	p.rwLocker.Unlock()
 
 	p.lastProcessingTime.Store(time.Now())
-
-	startTime := time.Now()
-	defer func() {
-		duration := time.Since(startTime)
-		p.lastProcessingDuration.Store(duration)
-	}()
-
-	if err := p.mailbox.Push(ctx); err != nil {
+	if err := mailbox.Push(ctx); err != nil {
 		p.logger.Warn(err)
 		p.handleError(ctx, err)
-		return
 	}
 }
 
@@ -1172,9 +1176,9 @@ func (p *pid) init(ctx context.Context) error {
 		return e
 	}
 
-	p.rwMutex.Lock()
+	p.rwLocker.Lock()
 	p.isRunning.Store(true)
-	p.rwMutex.Unlock()
+	p.rwLocker.Unlock()
 
 	p.logger.Info("Initialization process successfully completed.")
 	span.SetStatus(codes.Ok, "Init")
@@ -1191,7 +1195,7 @@ func (p *pid) init(ctx context.Context) error {
 
 // reset re-initializes the actor PID
 func (p *pid) reset() {
-	p.lastProcessingTime = atomic.Time{}
+	p.lastProcessingTime.Store(time.Time{})
 	p.passivateAfter.Store(DefaultPassivationTimeout)
 	p.replyTimeout.Store(DefaultReplyTimeout)
 	p.shutdownTimeout.Store(DefaultShutdownTimeout)
@@ -1350,8 +1354,8 @@ func (p *pid) passivationListener() {
 		return
 	}
 
-	p.stopMutex.Lock()
-	defer p.stopMutex.Unlock()
+	p.stopLocker.Lock()
+	defer p.stopLocker.Unlock()
 
 	p.logger.Infof("Passivation mode has been triggered for actor=%s...", p.ActorPath().String())
 
@@ -1376,33 +1380,33 @@ func (p *pid) passivationListener() {
 
 // setBehavior is a utility function that helps set the actor behavior
 func (p *pid) setBehavior(behavior Behavior) {
-	p.rwMutex.Lock()
+	p.rwLocker.Lock()
 	p.behaviorStack.Clear()
 	p.behaviorStack.Push(behavior)
-	p.rwMutex.Unlock()
+	p.rwLocker.Unlock()
 }
 
 // resetBehavior is a utility function resets the actor behavior
 func (p *pid) resetBehavior() {
-	p.rwMutex.Lock()
+	p.rwLocker.Lock()
 	p.behaviorStack.Clear()
 	p.behaviorStack.Push(p.Receive)
-	p.rwMutex.Unlock()
+	p.rwLocker.Unlock()
 }
 
 // setBehaviorStacked adds a behavior to the actor's behaviorStack
 func (p *pid) setBehaviorStacked(behavior Behavior) {
-	p.rwMutex.Lock()
+	p.rwLocker.Lock()
 	p.behaviorStack.Push(behavior)
-	p.rwMutex.Unlock()
+	p.rwLocker.Unlock()
 }
 
 // unsetBehaviorStacked sets the actor's behavior to the previous behavior
 // prior to setBehaviorStacked is called
 func (p *pid) unsetBehaviorStacked() {
-	p.rwMutex.Lock()
+	p.rwLocker.Lock()
 	p.behaviorStack.Pop()
-	p.rwMutex.Unlock()
+	p.rwLocker.Unlock()
 }
 
 // doStop stops the actor
@@ -1558,3 +1562,16 @@ func (p *pid) getRemoteServiceClient(host string, port int) (internalpbconnect.R
 		clientConnectionOptions...,
 	), nil
 }
+
+func (p *pid) getLastProcessingTime() time.Time {
+	p.processingTimeLocker.Lock()
+	processingTime := p.lastProcessingTime.Load()
+	p.processingTimeLocker.Unlock()
+	return processingTime
+}
+
+func (p *pid) setLastProcessingDuration(d time.Duration) {
+	p.processingTimeLocker.Lock()
+	p.lastProcessingDuration.Store(d)
+	p.processingTimeLocker.Unlock()
+}
diff --git a/actors/stash.go b/actors/stash.go
index 19873ae..778670f 100644
--- a/actors/stash.go
+++ b/actors/stash.go
@@ -30,8 +30,8 @@ func (p *pid) stash(ctx ReceiveContext) error {
 		return ErrStashBufferNotSet
 	}
 
-	p.stashMutex.Lock()
-	defer p.stashMutex.Unlock()
+	p.stashLocker.Lock()
+	defer p.stashLocker.Unlock()
 
 	return p.stashBuffer.Push(ctx)
 }
@@ -42,8 +42,8 @@ func (p *pid) unstash() error {
 		return ErrStashBufferNotSet
 	}
 
-	p.stashMutex.Lock()
-	defer p.stashMutex.Unlock()
+	p.stashLocker.Lock()
+	defer p.stashLocker.Unlock()
 
 	if !p.stashBuffer.IsEmpty() {
 		received, _ := p.stashBuffer.Pop()
@@ -59,8 +59,8 @@ func (p *pid) unstashAll() error {
 		return ErrStashBufferNotSet
 	}
 
-	p.stashMutex.Lock()
-	defer p.stashMutex.Unlock()
+	p.stashLocker.Lock()
+	defer p.stashLocker.Unlock()
 
 	for !p.stashBuffer.IsEmpty() {
 		received, _ := p.stashBuffer.Pop()


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index fcc0efe..5ae295d 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -1400,7 +1400,7 @@ func TestActorSystem(t *testing.T) {
 			cluster:        mockedCluster,
 			clusterEnabled: *atomic.NewBool(true),
 			telemetry:      telemetry.New(),
-			mutex:          sync.Mutex{},
+			locker:         sync.Mutex{},
 			tracer:         noop.NewTracerProvider().Tracer("testSystem"),
 			scheduler:      newScheduler(logger, time.Second, withSchedulerCluster(mockedCluster)),
 			clusterConfig:  NewClusterConfig(),
diff --git a/actors/pid_map_test.go b/actors/pid_map_test.go
index 6e2d42c..e859cdb 100644
--- a/actors/pid_map_test.go
+++ b/actors/pid_map_test.go
@@ -35,7 +35,7 @@ func TestPIDMap(t *testing.T) {
 	// create the actor path
 	actorPath := NewPath("Test", NewAddress("TestSys", "host", 444))
 	// create the PID
-	actorRef := &pid{actorPath: actorPath, rwMutex: &sync.RWMutex{}, stopMutex: &sync.Mutex{}}
+	actorRef := &pid{actorPath: actorPath, rwLocker: &sync.RWMutex{}, stopLocker: &sync.Mutex{}}
 	// create a new PID map
 	pidMap := newPIDMap(5)
 	// add to the map
diff --git a/actors/pid_test.go b/actors/pid_test.go
index 98dd7e3..94d8049 100644
--- a/actors/pid_test.go
+++ b/actors/pid_test.go
@@ -339,8 +339,8 @@ func TestRestart(t *testing.T) {
 	})
 	t.Run("noSender cannot be restarted", func(t *testing.T) {
 		pid := &pid{
-			tracer:  noop.NewTracerProvider().Tracer(""),
-			rwMutex: &sync.RWMutex{},
+			tracer:   noop.NewTracerProvider().Tracer(""),
+			rwLocker: &sync.RWMutex{},
 		}
 		err := pid.Restart(context.TODO())
 		assert.Error(t, err)
diff --git a/actors/stash_test.go b/actors/stash_test.go
index a1c374c..a988c5f 100644
--- a/actors/stash_test.go
+++ b/actors/stash_test.go
@@ -111,6 +111,7 @@ func TestStash(t *testing.T) {
 		opts := []pidOption{
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
+			withStash(0),
 		}
 
 		// create the actor path
@@ -136,6 +137,7 @@ func TestStash(t *testing.T) {
 		opts := []pidOption{
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
+			withStash(0),
 		}
 
 		// create the actor path
diff --git a/bench/bench_test.go b/bench/bench_test.go
index 8a8cf02..9d61631 100644
--- a/bench/bench_test.go
+++ b/bench/bench_test.go
@@ -59,7 +59,7 @@ func (p *Benchmarker) PostStop(context.Context) error {
 }
 
 func BenchmarkActor(b *testing.B) {
-	b.Run("tell(send only)", func(b *testing.B) {
+	b.Run("tell(send only) with parallelism", func(b *testing.B) {
 		ctx := context.TODO()
 
 		// create the actor system
@@ -89,7 +89,7 @@ func BenchmarkActor(b *testing.B) {
 		_ = pid.Shutdown(ctx)
 		_ = actorSystem.Stop(ctx)
 	})
-	b.Run("ask(send/reply)", func(b *testing.B) {
+	b.Run("ask(send/reply) with parallelism", func(b *testing.B) {
 		ctx := context.TODO()
 		// create the actor system
 		actorSystem, _ := actors.NewActorSystem("testSys",
@@ -118,4 +118,60 @@ func BenchmarkActor(b *testing.B) {
 		_ = pid.Shutdown(ctx)
 		_ = actorSystem.Stop(ctx)
 	})
+	b.Run("tell(send only)", func(b *testing.B) {
+		ctx := context.TODO()
+
+		// create the actor system
+		actorSystem, _ := actors.NewActorSystem("testSys",
+			actors.WithLogger(log.DiscardLogger),
+			actors.WithActorInitMaxRetries(1),
+			actors.WithMailboxSize(uint64(b.N)),
+			actors.WithReplyTimeout(receivingTimeout))
+
+		// start the actor system
+		_ = actorSystem.Start(ctx)
+
+		// define the benchmark actor
+		actor := &Benchmarker{}
+
+		// create the actor ref
+		pid, _ := actorSystem.Spawn(ctx, "test", actor)
+
+		b.ResetTimer() // Reset the benchmark timer
+		for i := 0; i < b.N; i++ {
+			// send a message to the actor
+			_ = actors.Tell(ctx, pid, &testspb.TestSend{})
+		}
+
+		_ = pid.Shutdown(ctx)
+		_ = actorSystem.Stop(ctx)
+	})
+	b.Run("ask(send/reply)", func(b *testing.B) {
+		ctx := context.TODO()
+		// create the actor system
+		actorSystem, _ := actors.NewActorSystem("testSys",
+			actors.WithLogger(log.DiscardLogger),
+			actors.WithActorInitMaxRetries(1),
+			actors.WithMailboxSize(uint64(b.N)),
+			actors.WithExpireActorAfter(5*time.Second),
+			actors.WithReplyTimeout(receivingTimeout))
+
+		// start the actor system
+		_ = actorSystem.Start(ctx)
+
+		// define the benchmark actor
+		actor := &Benchmarker{}
+
+		// create the actor ref
+		pid, _ := actorSystem.Spawn(ctx, "test", actor)
+
+		b.ResetTimer() // Reset the benchmark timer
+		for i := 0; i < b.N; i++ {
+			// send a message to the actor
+			_, _ = actors.Ask(ctx, pid, new(testspb.TestReply), receivingTimeout)
+		}
+
+		_ = pid.Shutdown(ctx)
+		_ = actorSystem.Stop(ctx)
+	})
 }