Repository: Tochemey/goakt
Pull request #559 (2024-12-13 11:06:59)

perf: change boundedmailbox implementation

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/559?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 `92.30769%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 87.13%. Comparing base [(`14dc9d1`)](https://app.codecov.io/gh/Tochemey/goakt/commit/14dc9d12cd8079e925f932da9970e8d96b23258c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) to head [(`ca61d96`)](https://app.codecov.io/gh/Tochemey/goakt/commit/ca61d96afcee5c58d22f66e8b38561ee9b9fad92?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).

| [Files with missing lines](https://app.codecov.io/gh/Tochemey/goakt/pull/559?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/bounded\_mailbox.go](https://app.codecov.io/gh/Tochemey/goakt/pull/559?src=pr&el=tree&filepath=actors%2Fbounded_mailbox.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL2JvdW5kZWRfbWFpbGJveC5nbw==) | 90.00% | [1 Missing :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/559?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     #559      +/-   ##
==========================================
- Coverage   87.14%   87.13%   -0.01%     
==========================================
  Files          54       54              
  Lines        5118     5093      -25     
==========================================
- Hits         4460     4438      -22     
+ Misses        519      517       -2     
+ Partials      139      138       -1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/559?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 33be01d..03267e5 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -147,6 +147,8 @@ type ActorSystem interface {
 	// Port returns the actor system node port.
 	// This is the bind port for remote communication
 	Port() int32
+	// Uptime returns the number of seconds since the actor system started
+	Uptime() int64
 	// handleRemoteAsk handles a synchronous message to another actor and expect a response.
 	// This block until a response is received or timed out.
 	handleRemoteAsk(ctx context.Context, to *PID, message proto.Message, timeout time.Duration) (response proto.Message, err error)
@@ -250,6 +252,7 @@ type actorSystem struct {
 	userGuardian   *PID
 	systemGuardian *PID
 	janitor        *PID
+	startedAt      *atomic.Int64
 }
 
 // enforce compilation error when all methods of the ActorSystem interface are not implemented
@@ -287,6 +290,7 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		port:                   0,
 		host:                   "127.0.0.1",
 		actors:                 newTree(),
+		startedAt:              atomic.NewInt64(0),
 	}
 
 	system.started.Store(false)
@@ -320,6 +324,14 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 	return system, nil
 }
 
+// Uptime returns the number of seconds since the actor system started
+func (x *actorSystem) Uptime() int64 {
+	if x.started.Load() {
+		return time.Now().Unix() - x.startedAt.Load()
+	}
+	return 0
+}
+
 // Host returns the actor system node host address
 // This is the bind address for remote communication
 func (x *actorSystem) Host() string {
@@ -728,7 +740,7 @@ func (x *actorSystem) Start(ctx context.Context) error {
 	}
 
 	x.scheduler.Start(ctx)
-
+	x.startedAt.Store(time.Now().Unix())
 	x.logger.Infof("%s started..:)", x.name)
 	return nil
 }
diff --git a/actors/bounded_mailbox.go b/actors/bounded_mailbox.go
index a470ad5..884349d 100644
--- a/actors/bounded_mailbox.go
+++ b/actors/bounded_mailbox.go
@@ -25,17 +25,13 @@
 package actors
 
 import (
-	"sync"
+	gods "github.com/Workiva/go-datastructures/queue"
 )
 
 // BoundedMailbox defines a bounded mailbox using ring buffer queue
 // This mailbox is thread-safe
 type BoundedMailbox struct {
-	buffer     []*ReceiveContext
-	head, tail int
-	len, cap   int
-	full       bool
-	lock       *sync.Mutex
+	underlying *gods.RingBuffer
 }
 
 // enforce compilation error
@@ -44,66 +40,38 @@ var _ Mailbox = (*BoundedMailbox)(nil)
 // NewBoundedMailbox creates a new instance BoundedMailbox
 func NewBoundedMailbox(cap int) *BoundedMailbox {
 	return &BoundedMailbox{
-		buffer: make([]*ReceiveContext, cap),
-		head:   0,
-		tail:   0,
-		len:    0,
-		cap:    cap,
-		lock:   new(sync.Mutex),
+		underlying: gods.NewRingBuffer(uint64(cap)),
 	}
 }
 
 // Enqueue places the given value in the mailbox
 // This will return an error when the mailbox is full
 func (mailbox *BoundedMailbox) Enqueue(msg *ReceiveContext) error {
-	if mailbox.isFull() {
-		return ErrFullMailbox
-	}
-
-	mailbox.lock.Lock()
-	mailbox.buffer[mailbox.tail] = msg
-	mailbox.tail = (mailbox.tail + 1) % mailbox.cap
-	mailbox.len++
-	mailbox.full = mailbox.head == mailbox.tail
-	mailbox.lock.Unlock()
-	return nil
+	return mailbox.underlying.Put(msg)
 }
 
 // Dequeue takes the mail from the mailbox
 // It returns nil when the mailbox is empty
 func (mailbox *BoundedMailbox) Dequeue() (msg *ReceiveContext) {
-	if mailbox.IsEmpty() {
-		return nil
+	if mailbox.underlying.Len() > 0 {
+		item, _ := mailbox.underlying.Get()
+		return item.(*ReceiveContext)
 	}
-
-	mailbox.lock.Lock()
-	item := mailbox.buffer[mailbox.head]
-	mailbox.full = false
-	mailbox.head = (mailbox.head + 1) % mailbox.cap
-	mailbox.len--
-	mailbox.lock.Unlock()
-	return item
+	return nil
 }
 
 // IsEmpty returns true when the mailbox is empty
 func (mailbox *BoundedMailbox) IsEmpty() bool {
-	mailbox.lock.Lock()
-	empty := mailbox.head == mailbox.tail && !mailbox.full
-	mailbox.lock.Unlock()
-	return empty
+	return mailbox.underlying.Len() == 0
 }
 
 // Len returns queue length
 func (mailbox *BoundedMailbox) Len() int64 {
-	mailbox.lock.Lock()
-	length := mailbox.len
-	mailbox.lock.Unlock()
-	return int64(length)
+	return int64(mailbox.underlying.Len())
 }
 
-func (mailbox *BoundedMailbox) isFull() bool {
-	mailbox.lock.Lock()
-	full := mailbox.full
-	mailbox.lock.Unlock()
-	return full
+// Dispose will dispose of this queue and free any blocked threads
+// in the Enqueue and/or Dequeue methods.
+func (mailbox *BoundedMailbox) Dispose() {
+	mailbox.underlying.Dispose()
 }
diff --git a/actors/mailbox.go b/actors/mailbox.go
index 110d8ba..de928ea 100644
--- a/actors/mailbox.go
+++ b/actors/mailbox.go
@@ -36,4 +36,7 @@ type Mailbox interface {
 	IsEmpty() bool
 	// Len returns the size of the mailbox
 	Len() int64
+	// Dispose will dispose of this queue and free any blocked threads
+	// in the Enqueue and/or Dequeue methods.
+	Dispose()
 }
diff --git a/actors/pid.go b/actors/pid.go
index 16f942f..a8b3202 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -143,6 +143,7 @@ type PID struct {
 	remoting *Remoting
 
 	goScheduler *goScheduler
+	startedAt   *atomic.Int64
 }
 
 // newPID creates a new pid
@@ -177,6 +178,7 @@ func newPID(ctx context.Context, address *address.Address, actor Actor, opts ...
 		remoting:              NewRemoting(),
 		goScheduler:           newGoScheduler(300),
 		supervisorStrategies:  newStrategiesMap(),
+		startedAt:             atomic.NewInt64(0),
 	}
 
 	pid.initMaxRetries.Store(DefaultInitMaxRetries)
@@ -214,9 +216,18 @@ func newPID(ctx context.Context, address *address.Address, actor Actor, opts ...
 	receiveContext.build(ctx, NoSender, pid, new(goaktpb.PostStart), true)
 	pid.doReceive(receiveContext)
 
+	pid.startedAt.Store(time.Now().Unix())
 	return pid, nil
 }
 
+// Uptime returns the number of seconds since the actor started
+func (pid *PID) Uptime() int64 {
+	if pid.IsRunning() {
+		return time.Now().Unix() - pid.startedAt.Load()
+	}
+	return 0
+}
+
 // ID is a convenient method that returns the actor unique identifier
 // An actor unique identifier is its address in the actor system.
 func (pid *PID) ID() string {
@@ -1268,9 +1279,12 @@ func (pid *PID) reset() {
 	pid.initTimeout.Store(DefaultInitTimeout)
 	pid.behaviorStack.Reset()
 	pid.processedCount.Store(0)
+	pid.restartCount.Store(0)
+	pid.startedAt.Store(0)
 	pid.stopping.Store(false)
 	pid.suspended.Store(false)
 	pid.supervisorStrategies.Reset()
+	pid.mailbox.Dispose()
 }
 
 // freeWatchers releases all the actors watching this actor
diff --git a/actors/unbounded_mailbox.go b/actors/unbounded_mailbox.go
index 340455e..08309f2 100644
--- a/actors/unbounded_mailbox.go
+++ b/actors/unbounded_mailbox.go
@@ -90,3 +90,7 @@ func (m *UnboundedMailbox) Len() int64 {
 func (m *UnboundedMailbox) IsEmpty() bool {
 	return atomic.LoadInt64(&m.length) == 0
 }
+
+// Dispose will dispose of this queue and free any blocked threads
+// in the Enqueue and/or Dequeue methods.
+func (m *UnboundedMailbox) Dispose() {}
diff --git a/actors/unbounded_priority_mailbox.go b/actors/unbounded_priority_mailbox.go
index 9ff5bb7..c041dec 100644
--- a/actors/unbounded_priority_mailbox.go
+++ b/actors/unbounded_priority_mailbox.go
@@ -132,3 +132,7 @@ func (q *UnboundedPriorityMailBox) IsEmpty() bool {
 func (q *UnboundedPriorityMailBox) Len() int64 {
 	return atomic.LoadInt64(&q.length)
 }
+
+// Dispose will dispose of this queue and free any blocked threads
+// in the Enqueue and/or Dequeue methods.
+func (q *UnboundedPriorityMailBox) Dispose() {}
diff --git a/go.mod b/go.mod
index 694e42b..2491775 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ go 1.22.0
 
 require (
 	connectrpc.com/connect v1.17.0
+	github.com/Workiva/go-datastructures v1.1.5
 	github.com/buraksezer/olric v0.5.6-0.20240925183822-6ca0e20256e0
 	github.com/deckarep/golang-set/v2 v2.7.0
 	github.com/flowchartsman/retry v1.2.0
diff --git a/go.sum b/go.sum
index b87b43d..0fc5be8 100644
--- a/go.sum
+++ b/go.sum
@@ -9,6 +9,8 @@ github.com/RoaringBitmap/roaring v1.6.0/go.mod h1:plvDsJQpxOC5bw8LRteu/MLWHsHez/
 github.com/RoaringBitmap/roaring v1.9.4 h1:yhEIoH4YezLYT04s1nHehNO64EKFTop/wBhxv2QzDdQ=
 github.com/RoaringBitmap/roaring v1.9.4/go.mod h1:6AXUsoIEzDTFFQCe1RbGA6uFONMhvejWj5rqITANK90=
 github.com/Sereal/Sereal/Go/sereal v0.0.0-20231009093132-b9187f1a92c6/go.mod h1:JwrycNnC8+sZPDyzM3MQ86LvaGzSpfxg885KOOwFRW4=
+github.com/Workiva/go-datastructures v1.1.5 h1:5YfhQ4ry7bZc2Mc7R0YZyYwpf5c6t1cEFvdAhd6Mkf4=
+github.com/Workiva/go-datastructures v1.1.5/go.mod h1:1yZL+zfsztete+ePzZz/Zb1/t5BnDuE2Ya2MMGhzP6A=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -205,6 +207,7 @@ github.com/onsi/gomega v1.33.1/go.mod h1:U4R44UsT+9eLIaYRB2a5qajjtQYn0hauxvRm16A
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
 github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
+github.com/philhofer/fwd v1.1.1/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/philhofer/fwd v1.1.2/go.mod h1:qkPdfjR2SIEbspLqpe1tO4n5yICnr2DY7mqEx2tUTP0=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -267,9 +270,11 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/redcon v1.6.2 h1:5qfvrrybgtO85jnhSravmkZyC0D+7WstbfCs3MmPhow=
 github.com/tidwall/redcon v1.6.2/go.mod h1:p5Wbsgeyi2VSTBWOcA5vRXrOb9arFTcU2+ZzFjqV75Y=
+github.com/tinylib/msgp v1.1.5/go.mod h1:eQsjooMTnV42mHu917E26IogZ2930nFyBQdofk10Udg=
 github.com/tinylib/msgp v1.1.8/go.mod h1:qkpG+2ldGg4xRFmx+jfTvZPxfGFhi64BcnL9vkCm/Tw=
 github.com/travisjeffery/go-dynaport v1.0.0 h1:m/qqf5AHgB96CMMSworIPyo1i7NZueRsnwdzdCJ8Ajw=
 github.com/travisjeffery/go-dynaport v1.0.0/go.mod h1:0LHuDS4QAx+mAc4ri3WkQdavgVoBIZ7cE9ob17KIAJk=
+github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31/go.mod h1:onvgF043R+lC5RZ8IT9rBXDaEDnpnw/Cl+HFiw+v/7Q=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
 github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
@@ -423,6 +428,7 @@ golang.org/x/tools v0.0.0-20190907020128-2ca718005c18/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191216052735-49a3e744a425/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20201022035929-9cf592e881e9/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index 117f03b..5bdfbb5 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -79,6 +79,7 @@ func TestActorSystem(t *testing.T) {
 		assert.Error(t, err)
 		assert.EqualError(t, err, ErrActorSystemNotStarted.Error())
 		assert.Nil(t, actorRef)
+		assert.Zero(t, sys.Uptime())
 	})
 	t.Run("With Spawn an actor when started", func(t *testing.T) {
 		ctx := context.TODO()
@@ -95,6 +96,8 @@ func TestActorSystem(t *testing.T) {
 		assert.NoError(t, err)
 		assert.NotNil(t, actorRef)
 
+		assert.NotZero(t, sys.Uptime())
+
 		// stop the actor after some time
 		lib.Pause(time.Second)
 		err = sys.Stop(ctx)
diff --git a/actors/bounded_mailbox_test.go b/actors/bounded_mailbox_test.go
index d5a8854..c9f8870 100644
--- a/actors/bounded_mailbox_test.go
+++ b/actors/bounded_mailbox_test.go
@@ -37,9 +37,6 @@ func TestBoundedMailbox(t *testing.T) {
 		require.NoError(t, mailbox.Enqueue(&ReceiveContext{}))
 	}
 	assert.False(t, mailbox.IsEmpty())
-	err := mailbox.Enqueue(&ReceiveContext{})
-	require.Error(t, err)
-	assert.EqualError(t, err, ErrFullMailbox.Error())
 	dequeue := mailbox.Dequeue()
 	require.NotNil(t, dequeue)
 	assert.EqualValues(t, 19, mailbox.Len())
@@ -49,4 +46,7 @@ func TestBoundedMailbox(t *testing.T) {
 		counter--
 	}
 	assert.True(t, mailbox.IsEmpty())
+	mailbox.Dispose()
+	err := mailbox.Enqueue(&ReceiveContext{})
+	require.Error(t, err)
 }
diff --git a/actors/pid_test.go b/actors/pid_test.go
index b842216..7a82068 100644
--- a/actors/pid_test.go
+++ b/actors/pid_test.go
@@ -283,11 +283,16 @@ func TestRestart(t *testing.T) {
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
 
+		lib.Pause(time.Second)
+
+		assert.NotZero(t, pid.Uptime())
+
 		// stop the actor
 		err = pid.Shutdown(ctx)
 		assert.NoError(t, err)
 
 		lib.Pause(time.Second)
+		assert.Zero(t, pid.Uptime())
 
 		// let us send a message to the actor
 		err = Tell(ctx, pid, new(testpb.TestSend))
@@ -298,6 +303,7 @@ func TestRestart(t *testing.T) {
 		err = pid.Restart(ctx)
 		assert.NoError(t, err)
 		assert.True(t, pid.IsRunning())
+		assert.NotZero(t, pid.Uptime())
 		// let us send 10 public to the actor
 		count := 10
 		for i := 0; i < count; i++ {
diff --git a/actors/unbounded_mailbox_test.go b/actors/unbounded_mailbox_test.go
index 6cd5de6..93e5cbc 100644
--- a/actors/unbounded_mailbox_test.go
+++ b/actors/unbounded_mailbox_test.go
@@ -50,6 +50,7 @@ func TestUnboundedMailbox(t *testing.T) {
 	assert.Equal(t, in1, out1)
 	assert.Equal(t, in2, out2)
 	assert.True(t, mailbox.IsEmpty())
+	mailbox.Dispose()
 }
 
 func TestUnboundedMailboxOneProducer(t *testing.T) {
diff --git a/actors/unbounded_priority_mailbox_test.go b/actors/unbounded_priority_mailbox_test.go
index 966fcda..b3b7850 100644
--- a/actors/unbounded_priority_mailbox_test.go
+++ b/actors/unbounded_priority_mailbox_test.go
@@ -69,6 +69,7 @@ func TestUnboundedPriorityMailBox(t *testing.T) {
 		require.True(t, ok)
 		require.EqualValues(t, 1, msg.GetPriority())
 		require.True(t, mailbox.IsEmpty())
+		mailbox.Dispose()
 	})
 	t.Run("With lowest priority mailbox", func(t *testing.T) {
 		priorityFunc := func(msg1, msg2 proto.Message) bool {
diff --git a/bench/benchmark_test.go b/bench/benchmark_test.go
index 55572e6..94da8a6 100644
--- a/bench/benchmark_test.go
+++ b/bench/benchmark_test.go
@@ -322,6 +322,45 @@ func BenchmarkActor(b *testing.B) {
 		messagesPerSec := float64(atomic.LoadInt64(&counter)) / b.Elapsed().Seconds()
 		b.ReportMetric(messagesPerSec, "messages/sec")
 
+		_ = actorSystem.Stop(ctx)
+	})
+	b.Run("Ask(bounded mailbox)", func(b *testing.B) {
+		ctx := context.TODO()
+
+		// create the actor system
+		actorSystem, _ := actors.NewActorSystem("bench",
+			actors.WithLogger(log.DiscardLogger),
+			actors.WithActorInitMaxRetries(1))
+
+		// start the actor system
+		_ = actorSystem.Start(ctx)
+
+		// wait for system to start properly
+		lib.Pause(1 * time.Second)
+
+		// create the actors
+		sender, _ := actorSystem.Spawn(ctx, "sender", new(Benchmarker), actors.WithMailbox(actors.NewBoundedMailbox(b.N)))
+		receiver, _ := actorSystem.Spawn(ctx, "receiver", new(Benchmarker), actors.WithMailbox(actors.NewBoundedMailbox(b.N)))
+
+		// wait for actors to start properly
+		lib.Pause(1 * time.Second)
+
+		var counter int64
+		b.ResetTimer()
+		b.ReportAllocs()
+		b.RunParallel(func(pb *testing.PB) {
+			for pb.Next() {
+				if _, err := sender.Ask(ctx, receiver, new(benchmarkpb.BenchRequest), receivingTimeout); err != nil {
+					b.Fatal(err)
+				}
+				atomic.AddInt64(&counter, 1)
+			}
+		})
+		b.StopTimer()
+
+		messagesPerSec := float64(atomic.LoadInt64(&counter)) / b.Elapsed().Seconds()
+		b.ReportMetric(messagesPerSec, "messages/sec")
+
 		_ = actorSystem.Stop(ctx)
 	})
 }