Repository: Tochemey/goakt
Pull request #403 (2024-07-29 21:39:40)

perf: remove custom mailbox and make use of channel instead of a wrapper


Patch:
diff --git a/README.md b/README.md
index 9bea583..19453f1 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,6 @@ The project adheres to [Semantic Versioning](https://semver.org) and [Convention
   - [Passivation](#passivation)
   - [Actor System](#actor-system)
   - [Behaviors](#behaviors)
-  - [Mailbox](#mailbox)
   - [Router](#router)
   - [Events Stream](#events-stream)
     - [Supported events](#supported-events)
@@ -171,10 +170,6 @@ To change the behavior, call the following methods on the [ReceiveContext interf
 - `BecomeStacked` - sets a new behavior to the actor to the top of the behavior stack, while maintaining the previous ones.
 - `UnBecomeStacked()` - sets the actor behavior to the previous behavior before `BecomeStacked()` was called. This only works with `BecomeStacked()`.
 
-### Mailbox
-
-Once can implement a custom mailbox. See [Mailbox](./actors/mailbox.go). The default mailbox makes use of buffered channels.
-
 ### Router
 
 Routers help send the same type of message to a set of actors to be processed in parallel depending upon
diff --git a/actors/actor_system.go b/actors/actor_system.go
index 787a825..9cd42df 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -203,8 +203,6 @@ type actorSystem struct {
 	locker sync.Mutex
 	// specifies actors mailbox size
 	mailboxSize uint64
-	// specifies the mailbox to use for the actors
-	mailbox Mailbox
 	// specifies the stash capacity
 	stashCapacity uint64
 
@@ -261,6 +259,7 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		locker:                 sync.Mutex{},
 		shutdownTimeout:        DefaultShutdownTimeout,
 		mailboxSize:            DefaultMailboxSize,
+		stashCapacity:          DefaultStashCapacity,
 		stopGC:                 make(chan types.Unit, 1),
 		gcInterval:             DefaultGCInterval,
 		eventsStream:           eventstream.New(),
@@ -1506,12 +1505,6 @@ func (x *actorSystem) configPID(ctx context.Context, name string, actor Actor) (
 		actorPath = NewPath(name, NewAddress(x.name, x.remotingHost, int(x.remotingPort)))
 	}
 
-	var mailbox Mailbox
-	if x.mailbox != nil {
-		// always create a fresh copy of provided mailbox for every new PID
-		mailbox = x.mailbox.Clone()
-	}
-
 	// define the pid options
 	// pid inherit the actor system settings defined during instantiation
 	pidOpts := []pidOption{
@@ -1521,7 +1514,6 @@ func (x *actorSystem) configPID(ctx context.Context, name string, actor Actor) (
 		withActorSystem(x),
 		withSupervisorDirective(x.supervisorDirective),
 		withMailboxSize(x.mailboxSize),
-		withMailbox(mailbox),
 		withStash(x.stashCapacity),
 		withEventsStream(x.eventsStream),
 		withInitTimeout(x.actorInitTimeout),
diff --git a/actors/mailbox.go b/actors/mailbox.go
deleted file mode 100644
index 47e0306..0000000
--- a/actors/mailbox.go
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * MIT License
- *
- * Copyright (c) 2022-2024 Tochemey
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all
- * copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package actors
-
-import (
-	"sync"
-
-	"go.uber.org/atomic"
-)
-
-// Mailbox defines the actor mailbox.
-// Any implementation should be a thread-safe FIFO
-type Mailbox interface {
-	// Push pushes a message into the mailbox. This returns an error
-	// when the box is full
-	Push(msg ReceiveContext) error
-	// Pop fetches a message from the mailbox
-	Pop() (msg ReceiveContext, err error)
-	// Iterator returns a channel that can be used to iterate over the mailbox
-	Iterator() <-chan ReceiveContext
-	// IsEmpty returns true when the mailbox is empty
-	IsEmpty() bool
-	// IsFull returns true when the mailbox is full
-	IsFull() bool
-	// Size returns the size of the buffer atomically
-	Size() uint64
-	// Reset resets the mailbox
-	Reset()
-	// Clone clones the current mailbox and returns a new Mailbox with reset settings
-	Clone() Mailbox
-	// Capacity returns the mailbox capacity atomically
-	Capacity() uint64
-}
-
-// receiveContextBuffer is the actor default inbox
-type receiveContextBuffer struct {
-	// specifies the number of messages to stash
-	capacity *atomic.Uint64
-	buffer   chan ReceiveContext
-	mu       *sync.RWMutex
-}
-
-// newReceiveContextBuffer creates a Mailbox with a fixed capacity
-func newReceiveContextBuffer(capacity uint64) Mailbox {
-	return &receiveContextBuffer{
-		capacity: atomic.NewUint64(capacity),
-		buffer:   make(chan ReceiveContext, capacity),
-		mu:       &sync.RWMutex{},
-	}
-}
-
-// enforce compilation error
-var _ Mailbox = &receiveContextBuffer{}
-
-// Push pushes a message into the mailbox. This returns an error
-// when the box is full
-func (x *receiveContextBuffer) Push(msg ReceiveContext) error {
-	x.mu.Lock()
-	defer x.mu.Unlock()
-	select {
-	case x.buffer <- msg:
-		return nil
-	default:
-		return ErrFullMailbox
-	}
-}
-
-// Pop fetches a message from the mailbox
-func (x *receiveContextBuffer) Pop() (msg ReceiveContext, err error) {
-	x.mu.RLock()
-	defer x.mu.RUnlock()
-	select {
-	case msg := <-x.buffer:
-		return msg, nil
-	default:
-		return nil, ErrEmptyMailbox
-	}
-}
-
-// Iterator returns a channel that can be used to iterate over the mailbox
-func (x *receiveContextBuffer) Iterator() <-chan ReceiveContext {
-	x.mu.RLock()
-	defer x.mu.RUnlock()
-	return x.buffer
-}
-
-// IsEmpty returns true when the buffer is empty
-func (x *receiveContextBuffer) IsEmpty() bool {
-	x.mu.RLock()
-	defer x.mu.RUnlock()
-	return len(x.buffer) == 0
-}
-
-// Size returns the size of the buffer
-func (x *receiveContextBuffer) Size() uint64 {
-	return uint64(len(x.buffer))
-}
-
-// Clone clones the current mailbox and returns a new Mailbox with reset settings
-func (x *receiveContextBuffer) Clone() Mailbox {
-	return &receiveContextBuffer{
-		capacity: x.capacity,
-		buffer:   make(chan ReceiveContext, x.capacity.Load()),
-		mu:       &sync.RWMutex{},
-	}
-}
-
-// Reset resets the mailbox
-func (x *receiveContextBuffer) Reset() {
-	x.mu.Lock()
-	defer x.mu.Unlock()
-	close(x.buffer)
-	x.buffer = make(chan ReceiveContext, x.capacity.Load())
-}
-
-// IsFull returns true when the mailbox is full
-func (x *receiveContextBuffer) IsFull() bool {
-	return x.Size() >= x.capacity.Load()
-}
-
-// Capacity implements Mailbox.
-func (x *receiveContextBuffer) Capacity() uint64 {
-	return x.capacity.Load()
-}
diff --git a/actors/option.go b/actors/option.go
index 59df6f8..0131e02 100644
--- a/actors/option.go
+++ b/actors/option.go
@@ -152,13 +152,6 @@ func WithMailboxSize(size uint64) Option {
 	})
 }
 
-// WithMailbox sets the custom mailbox used by the actors in the actor system
-func WithMailbox(mailbox Mailbox) Option {
-	return OptionFunc(func(a *actorSystem) {
-		a.mailbox = mailbox
-	})
-}
-
 // WithStash sets the stash buffer size
 func WithStash(capacity uint64) Option {
 	return OptionFunc(func(a *actorSystem) {
diff --git a/actors/pid.go b/actors/pid.go
index 2790d55..7c6d127 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -223,7 +223,7 @@ type pid struct {
 	shutdownTimeout atomic.Duration
 
 	// specifies the actor mailbox
-	mailbox     Mailbox
+	mailbox     chan ReceiveContext
 	mailboxSize uint64
 
 	// receives a shutdown signal. Once the signal is received
@@ -249,9 +249,9 @@ type pid struct {
 	// specifies the last processing message duration
 	lastProcessingDuration atomic.Duration
 
-	// rwLocker that helps synchronize the pid in a concurrent environment
+	// fieldsLocker that helps synchronize the pid in a concurrent environment
 	// this helps protect the pid fields accessibility
-	rwLocker *sync.RWMutex
+	fieldsLocker *sync.RWMutex
 
 	stopLocker           *sync.Mutex
 	processingTimeLocker *sync.Mutex
@@ -269,7 +269,7 @@ type pid struct {
 	behaviorStack *behaviorStack
 
 	// stash settings
-	stashBuffer   Mailbox
+	stashBuffer   chan ReceiveContext
 	stashCapacity atomic.Uint64
 	stashLocker   *sync.Mutex
 
@@ -314,7 +314,7 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 		watchedList:          newPIDMap(10),
 		telemetry:            telemetry.New(),
 		actorPath:            actorPath,
-		rwLocker:             &sync.RWMutex{},
+		fieldsLocker:         &sync.RWMutex{},
 		stopLocker:           &sync.Mutex{},
 		httpClient:           http.NewClient(),
 		mailbox:              nil,
@@ -342,12 +342,9 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 		opt(p)
 	}
 
-	if p.mailbox == nil {
-		p.mailbox = newReceiveContextBuffer(p.mailboxSize)
-	}
-
+	p.mailbox = make(chan ReceiveContext, p.mailboxSize)
 	if p.stashCapacity.Load() > 0 {
-		p.stashBuffer = newReceiveContextBuffer(p.stashCapacity.Load())
+		p.stashBuffer = make(chan ReceiveContext, p.stashCapacity.Load())
 	}
 
 	behaviorStack := newBehaviorStack()
@@ -416,9 +413,9 @@ func (x *pid) Child(name string) (PID, error) {
 // Parents returns the list of all direct parents of a given actor.
 // Only alive actors are included in the list or an empty list is returned
 func (x *pid) Parents() []PID {
-	x.rwLocker.Lock()
+	x.fieldsLocker.Lock()
 	watchers := x.watchersList
-	x.rwLocker.Unlock()
+	x.fieldsLocker.Unlock()
 	var parents []PID
 	if watchers.Len() > 0 {
 		for item := range watchers.Iter() {
@@ -437,9 +434,9 @@ func (x *pid) Parents() []PID {
 // Children returns the list of all the direct children of the given actor
 // Only alive actors are included in the list or an empty list is returned
 func (x *pid) Children() []PID {
-	x.rwLocker.RLock()
+	x.fieldsLocker.RLock()
 	children := x.children.pids()
-	x.rwLocker.RUnlock()
+	x.fieldsLocker.RUnlock()
 
 	cids := make([]PID, 0, len(children))
 	for _, child := range children {
@@ -469,9 +466,9 @@ func (x *pid) Stop(ctx context.Context, cid PID) error {
 		return ErrUndefinedActor
 	}
 
-	x.rwLocker.RLock()
+	x.fieldsLocker.RLock()
 	children := x.children
-	x.rwLocker.RUnlock()
+	x.fieldsLocker.RUnlock()
 
 	if cid, ok := children.get(cid.ActorPath()); ok {
 		desc := fmt.Sprintf("child.[%s] Shutdown", cid.ActorPath())
@@ -499,17 +496,17 @@ func (x *pid) IsRunning() bool {
 
 // ActorSystem returns the actor system
 func (x *pid) ActorSystem() ActorSystem {
-	x.rwLocker.RLock()
+	x.fieldsLocker.RLock()
 	sys := x.system
-	x.rwLocker.RUnlock()
+	x.fieldsLocker.RUnlock()
 	return sys
 }
 
 // ActorPath returns the path of the actor
 func (x *pid) ActorPath() *Path {
-	x.rwLocker.RLock()
+	x.fieldsLocker.RLock()
 	path := x.actorPath
-	x.rwLocker.RUnlock()
+	x.fieldsLocker.RUnlock()
 	return path
 }
 
@@ -545,7 +542,6 @@ func (x *pid) Restart(ctx context.Context) error {
 		ticker.Stop()
 	}
 
-	x.mailbox.Reset()
 	x.resetBehavior()
 	if err := x.init(spanCtx); err != nil {
 		return err
@@ -582,15 +578,15 @@ func (x *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 
 	childActorPath := NewPath(name, x.ActorPath().Address()).WithParent(x.ActorPath())
 
-	x.rwLocker.RLock()
+	x.fieldsLocker.RLock()
 	children := x.children
-	x.rwLocker.RUnlock()
+	x.fieldsLocker.RUnlock()
 
 	if cid, ok := children.get(childActorPath); ok {
 		return cid, nil
 	}
 
-	x.rwLocker.RLock()
+	x.fieldsLocker.RLock()
 
 	// create the child actor options child inherit parent's options
 	opts := []pidOption{
@@ -602,7 +598,6 @@ func (x *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 		withSupervisorDirective(x.supervisorDirective),
 		withMailboxSize(x.mailboxSize),
 		withStash(x.stashCapacity.Load()),
-		withMailbox(x.mailbox.Clone()),
 		withEventsStream(x.eventsStream),
 		withInitTimeout(x.initTimeout.Load()),
 		withShutdownTimeout(x.shutdownTimeout.Load()),
@@ -625,14 +620,14 @@ func (x *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 	if err != nil {
 		span.SetStatus(codes.Error, "SpawnChild")
 		span.RecordError(err)
-		x.rwLocker.RUnlock()
+		x.fieldsLocker.RUnlock()
 		return nil, err
 	}
 
 	x.children.set(cid)
 	eventsStream := x.eventsStream
 
-	x.rwLocker.RUnlock()
+	x.fieldsLocker.RUnlock()
 	span.SetStatus(codes.Ok, "SpawnChild")
 
 	x.Watch(cid)
@@ -658,7 +653,7 @@ func (x *pid) StashSize() uint64 {
 	if x.stashBuffer == nil {
 		return 0
 	}
-	return x.stashBuffer.Size()
+	return uint64(len(x.stashBuffer))
 }
 
 // PipeTo processes a long-running task and pipes the result to the provided actor.
@@ -1243,15 +1238,10 @@ func (x *pid) watchees() *pidMap {
 
 // doReceive pushes a given message to the actor receiveContextBuffer
 func (x *pid) doReceive(ctx ReceiveContext) {
-	x.rwLocker.Lock()
-	mailbox := x.mailbox
-	x.rwLocker.Unlock()
-
+	x.fieldsLocker.Lock()
 	x.lastProcessingTime.Store(time.Now())
-	if err := mailbox.Push(ctx); err != nil {
-		x.logger.Warn(err)
-		x.handleError(ctx, err)
-	}
+	x.mailbox <- ctx
+	x.fieldsLocker.Unlock()
 }
 
 // init initializes the given actor and init processing messages
@@ -1275,9 +1265,9 @@ func (x *pid) init(ctx context.Context) error {
 		return e
 	}
 
-	x.rwLocker.Lock()
+	x.fieldsLocker.Lock()
 	x.isRunning.Store(true)
-	x.rwLocker.Unlock()
+	x.fieldsLocker.Unlock()
 
 	x.logger.Info("Initialization process successfully completed.")
 	span.SetStatus(codes.Ok, "Init")
@@ -1304,7 +1294,6 @@ func (x *pid) reset() {
 	x.children = newPIDMap(10)
 	x.watchersList = slices.NewSafe[*watcher]()
 	x.telemetry = telemetry.New()
-	x.mailbox.Reset()
 	x.resetBehavior()
 	if x.metricEnabled.Load() {
 		if err := x.registerMetrics(); err != nil {
@@ -1371,7 +1360,7 @@ func (x *pid) receive() {
 		select {
 		case <-x.shutdownSignal:
 			return
-		case received, ok := <-x.mailbox.Iterator():
+		case received, ok := <-x.mailbox:
 			// break out of the loop when the channel is closed
 			if !ok {
 				return
@@ -1397,8 +1386,13 @@ func (x *pid) handleReceived(received ReceiveContext) {
 			}
 		}
 	}()
+
+	x.fieldsLocker.Lock()
+	behaviorStack := *x.behaviorStack
+	x.fieldsLocker.Unlock()
+
 	// send the message to the current actor behavior
-	if behavior, ok := x.behaviorStack.Peek(); ok {
+	if behavior, ok := behaviorStack.Peek(); ok {
 		behavior(received)
 	}
 }
@@ -1462,32 +1456,32 @@ func (x *pid) passivationListener() {
 
 // setBehavior is a utility function that helps set the actor behavior
 func (x *pid) setBehavior(behavior Behavior) {
-	x.rwLocker.Lock()
+	x.fieldsLocker.Lock()
 	x.behaviorStack.Clear()
 	x.behaviorStack.Push(behavior)
-	x.rwLocker.Unlock()
+	x.fieldsLocker.Unlock()
 }
 
 // resetBehavior is a utility function resets the actor behavior
 func (x *pid) resetBehavior() {
-	x.rwLocker.Lock()
+	x.fieldsLocker.Lock()
 	x.behaviorStack.Push(x.Receive)
-	x.rwLocker.Unlock()
+	x.fieldsLocker.Unlock()
 }
 
 // setBehaviorStacked adds a behavior to the actor's behaviorStack
 func (x *pid) setBehaviorStacked(behavior Behavior) {
-	x.rwLocker.Lock()
+	x.fieldsLocker.Lock()
 	x.behaviorStack.Push(behavior)
-	x.rwLocker.Unlock()
+	x.fieldsLocker.Unlock()
 }
 
 // unsetBehaviorStacked sets the actor's behavior to the previous behavior
 // prior to setBehaviorStacked is called
 func (x *pid) unsetBehaviorStacked() {
-	x.rwLocker.Lock()
+	x.fieldsLocker.Lock()
 	x.behaviorStack.Pop()
-	x.rwLocker.Unlock()
+	x.fieldsLocker.Unlock()
 }
 
 // doStop stops the actor
@@ -1498,7 +1492,7 @@ func (x *pid) doStop(ctx context.Context) error {
 	x.isRunning.Store(false)
 
 	// TODO: just signal stash processing done and ignore the messages or process them
-	for x.stashBuffer != nil && !x.stashBuffer.IsEmpty() {
+	if x.stashBuffer != nil {
 		if err := x.unstashAll(); err != nil {
 			x.logger.Errorf("actor=(%s) failed to unstash messages", x.ActorPath().String())
 			span.SetStatus(codes.Error, "doStop")
@@ -1518,7 +1512,7 @@ func (x *pid) doStop(ctx context.Context) error {
 		for {
 			select {
 			case <-ticker.C:
-				if x.mailbox.IsEmpty() {
+				if len(x.mailbox) == 0 {
 					close(tickerStopSig)
 					return
 				}
diff --git a/actors/pid_option.go b/actors/pid_option.go
index 85389e1..021c661 100644
--- a/actors/pid_option.go
+++ b/actors/pid_option.go
@@ -107,13 +107,6 @@ func withMailboxSize(size uint64) pidOption {
 	}
 }
 
-// withMailbox sets the custom actor receiveContextBuffer
-func withMailbox(box Mailbox) pidOption {
-	return func(pid *pid) {
-		pid.mailbox = box
-	}
-}
-
 // withStash sets the actor's stash buffer
 func withStash(capacity uint64) pidOption {
 	return func(pid *pid) {
diff --git a/actors/stash.go b/actors/stash.go
index 78d2751..7985d33 100644
--- a/actors/stash.go
+++ b/actors/stash.go
@@ -24,6 +24,8 @@
 
 package actors
 
+import "errors"
+
 // stash adds the current message to the stash buffer
 func (x *pid) stash(ctx ReceiveContext) error {
 	if x.stashBuffer == nil {
@@ -31,9 +33,9 @@ func (x *pid) stash(ctx ReceiveContext) error {
 	}
 
 	x.stashLocker.Lock()
-	defer x.stashLocker.Unlock()
-
-	return x.stashBuffer.Push(ctx)
+	x.stashBuffer <- ctx
+	x.stashLocker.Unlock()
+	return nil
 }
 
 // unstash unstashes the oldest message in the stash and prepends to the mailbox
@@ -44,11 +46,11 @@ func (x *pid) unstash() error {
 
 	x.stashLocker.Lock()
 	defer x.stashLocker.Unlock()
-
-	if !x.stashBuffer.IsEmpty() {
-		received, _ := x.stashBuffer.Pop()
-		x.doReceive(received)
+	received, ok := <-x.stashBuffer
+	if !ok {
+		return errors.New("stash buffer may be closed")
 	}
+	x.doReceive(received)
 	return nil
 }
 
@@ -62,8 +64,11 @@ func (x *pid) unstashAll() error {
 	x.stashLocker.Lock()
 	defer x.stashLocker.Unlock()
 
-	for !x.stashBuffer.IsEmpty() {
-		received, _ := x.stashBuffer.Pop()
+	for len(x.stashBuffer) > 0 {
+		received, ok := <-x.stashBuffer
+		if !ok {
+			return errors.New("stash buffer may be closed")
+		}
 		x.doReceive(received)
 	}
 


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index 036d30b..c2fd910 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -89,7 +89,7 @@ func TestActorSystem(t *testing.T) {
 	})
 	t.Run("With Spawn an actor when started", func(t *testing.T) {
 		ctx := context.TODO()
-		sys, _ := NewActorSystem("testSys", WithLogger(log.DefaultLogger), WithMailbox(newReceiveContextBuffer(100)))
+		sys, _ := NewActorSystem("testSys", WithLogger(log.DefaultLogger), WithMailboxSize(100))
 
 		// start the actor system
 		err := sys.Start(ctx)
@@ -1175,7 +1175,7 @@ func TestActorSystem(t *testing.T) {
 			WithLogger(logger),
 			WithReplyTimeout(time.Minute),
 			WithRemoting(host, int32(remotingPort)),
-			WithMailbox(newReceiveContextBuffer(100)),
+			WithMailboxSize(100),
 			WithClustering(provider, 9, 1, gossipPort, clusterPort, new(testActor)))
 		require.NoError(t, err)
 
diff --git a/actors/mailbox_test.go b/actors/mailbox_test.go
deleted file mode 100644
index bda4ca0..0000000
--- a/actors/mailbox_test.go
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * MIT License
- *
- * Copyright (c) 2022-2024 Tochemey
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all
- * copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package actors
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestMailbox(t *testing.T) {
-	t.Run("With happy path Push", func(t *testing.T) {
-		mailbox := newReceiveContextBuffer(10)
-		assert.True(t, mailbox.IsEmpty())
-		err := mailbox.Push(new(receiveContext))
-		assert.NoError(t, err)
-		assert.EqualValues(t, 1, mailbox.Size())
-		assert.EqualValues(t, 10, mailbox.Capacity())
-	})
-	t.Run("With Push when buffer is full", func(t *testing.T) {
-		mailbox := newReceiveContextBuffer(10)
-		assert.True(t, mailbox.IsEmpty())
-		for i := 0; i < 10; i++ {
-			err := mailbox.Push(new(receiveContext))
-			assert.NoError(t, err)
-		}
-		assert.EqualValues(t, 10, mailbox.Size())
-		// push another message to the buffer
-		err := mailbox.Push(new(receiveContext))
-		assert.Error(t, err)
-		assert.EqualError(t, err, ErrFullMailbox.Error())
-		assert.False(t, mailbox.IsEmpty())
-		assert.True(t, mailbox.IsFull())
-	})
-	t.Run("With happy path Pop", func(t *testing.T) {
-		mailbox := newReceiveContextBuffer(10)
-		assert.True(t, mailbox.IsEmpty())
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-
-		popped, err := mailbox.Pop()
-		assert.NoError(t, err)
-		assert.NotNil(t, popped)
-		assert.EqualValues(t, 3, mailbox.Size())
-	})
-
-	t.Run("With Clone", func(t *testing.T) {
-		mailbox := newReceiveContextBuffer(10)
-		assert.True(t, mailbox.IsEmpty())
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-
-		popped, err := mailbox.Pop()
-		assert.NoError(t, err)
-		assert.NotNil(t, popped)
-		assert.EqualValues(t, 3, mailbox.Size())
-
-		cloned := mailbox.Clone()
-		assert.True(t, cloned.IsEmpty())
-		assert.False(t, cloned.IsFull())
-	})
-	t.Run("With Pop when buffer is empty", func(t *testing.T) {
-		mailbox := newReceiveContextBuffer(10)
-		assert.True(t, mailbox.IsEmpty())
-
-		popped, err := mailbox.Pop()
-		assert.Error(t, err)
-		assert.EqualError(t, err, ErrEmptyMailbox.Error())
-		assert.Nil(t, popped)
-	})
-	t.Run("With Reset", func(t *testing.T) {
-		mailbox := newReceiveContextBuffer(10)
-		assert.True(t, mailbox.IsEmpty())
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-		assert.NoError(t, mailbox.Push(new(receiveContext)))
-
-		popped, err := mailbox.Pop()
-		assert.NoError(t, err)
-		assert.NotNil(t, popped)
-		assert.EqualValues(t, 3, mailbox.Size())
-		assert.False(t, mailbox.IsEmpty())
-
-		mailbox.Reset()
-		assert.True(t, mailbox.IsEmpty())
-		assert.False(t, mailbox.IsFull())
-	})
-}
diff --git a/actors/option_test.go b/actors/option_test.go
index 42900d2..3f1f808 100644
--- a/actors/option_test.go
+++ b/actors/option_test.go
@@ -38,7 +38,6 @@ import (
 
 func TestOption(t *testing.T) {
 	tel := telemetry.New()
-	mailbox := newReceiveContextBuffer(10)
 	resumeDirective := NewResumeDirective()
 	var atomicTrue atomic.Bool
 	atomicTrue.Store(true)
@@ -99,11 +98,6 @@ func TestOption(t *testing.T) {
 			option:   WithMailboxSize(10),
 			expected: actorSystem{mailboxSize: 10},
 		},
-		{
-			name:     "WithMailbox",
-			option:   WithMailbox(mailbox),
-			expected: actorSystem{mailbox: mailbox},
-		},
 		{
 			name:     "WithStash",
 			option:   WithStash(10),
diff --git a/actors/pid_map_test.go b/actors/pid_map_test.go
index e859cdb..68b9b01 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, rwLocker: &sync.RWMutex{}, stopLocker: &sync.Mutex{}}
+	actorRef := &pid{actorPath: actorPath, fieldsLocker: &sync.RWMutex{}, stopLocker: &sync.Mutex{}}
 	// create a new PID map
 	pidMap := newPIDMap(5)
 	// add to the map
diff --git a/actors/pid_option_test.go b/actors/pid_option_test.go
index 243a9f2..7bddaf9 100644
--- a/actors/pid_option_test.go
+++ b/actors/pid_option_test.go
@@ -36,7 +36,6 @@ import (
 )
 
 func TestPIDOptions(t *testing.T) {
-	mailbox := newReceiveContextBuffer(10)
 	resumeDirective := NewResumeDirective()
 	var (
 		atomicDuration   atomic.Duration
@@ -97,11 +96,6 @@ func TestPIDOptions(t *testing.T) {
 			option:   withMailboxSize(10),
 			expected: &pid{mailboxSize: 10},
 		},
-		{
-			name:     "WithMailbox",
-			option:   withMailbox(mailbox),
-			expected: &pid{mailbox: mailbox},
-		},
 		{
 			name:     "WithStash",
 			option:   withStash(10),
diff --git a/actors/pid_test.go b/actors/pid_test.go
index 6644c60..1708a31 100644
--- a/actors/pid_test.go
+++ b/actors/pid_test.go
@@ -341,8 +341,8 @@ func TestRestart(t *testing.T) {
 	})
 	t.Run("noSender cannot be restarted", func(t *testing.T) {
 		pid := &pid{
-			tracer:   noop.NewTracerProvider().Tracer(""),
-			rwLocker: &sync.RWMutex{},
+			tracer:       noop.NewTracerProvider().Tracer(""),
+			fieldsLocker: &sync.RWMutex{},
 		}
 		err := pid.Restart(context.TODO())
 		assert.Error(t, err)
diff --git a/bench/bench_test.go b/bench/bench_test.go
index e10bc28..71a727a 100644
--- a/bench/bench_test.go
+++ b/bench/bench_test.go
@@ -89,6 +89,38 @@ func BenchmarkActor(b *testing.B) {
 		_ = pid.Shutdown(ctx)
 		_ = actorSystem.Stop(ctx)
 	})
+	b.Run("tell without passivation", 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.WithSupervisorDirective(actors.NewStopDirective()),
+			actors.WithPassivationDisabled(),
+			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)
+
+		runParallel(b, func(pb *testing.PB) {
+			for pb.Next() {
+				// send a message to the actor
+				_ = actors.Tell(ctx, pid, &testspb.TestSend{})
+			}
+		})
+
+		_ = pid.Shutdown(ctx)
+		_ = actorSystem.Stop(ctx)
+	})
+
 	b.Run("ask", func(b *testing.B) {
 		ctx := context.TODO()
 		// create the actor system
@@ -114,6 +146,34 @@ func BenchmarkActor(b *testing.B) {
 			}
 		})
 
+		_ = pid.Shutdown(ctx)
+		_ = actorSystem.Stop(ctx)
+	})
+	b.Run("ask without passivation", 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.WithPassivationDisabled(),
+			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)
+		runParallel(b, func(pb *testing.PB) {
+			for pb.Next() {
+				// send a message to the actor
+				_, _ = actors.Ask(ctx, pid, new(testspb.TestReply), receivingTimeout)
+			}
+		})
+
 		_ = pid.Shutdown(ctx)
 		_ = actorSystem.Stop(ctx)
 	})