This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 9afe0e9109 fix(artery): distribute ActorSelection messages across
outbound lanes by target path (#3092)
9afe0e9109 is described below
commit 9afe0e910938fd932133086ecc3e8f41855ff5ff
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jun 24 21:09:57 2026 +0800
fix(artery): distribute ActorSelection messages across outbound lanes by
target path (#3092)
* fix(artery): distribute ActorSelection messages across outbound lanes by
target path
Motivation:
With multi-lane artery config (outbound-lanes > 1), all ActorSelection
messages were routed to the same outbound queue because selectQueue used
the anchor's UID (root guardian, always 0) as the distribution key:
math.abs(0 % N) = 0 for any N. This concentrated all ActorSelection
traffic on a single lane, creating a throughput bottleneck.
Modification:
Handle ActorSelectionMessage in a dedicated case that distributes across
lanes based on the target path elements hash instead of the anchor's UID.
PriorityMessage ActorSelection (cluster heartbeats) continues to use the
control queue. Uses (hash & Int.MaxValue) to guard against
Integer.MIN_VALUE producing a negative queue index.
Result:
ActorSelection messages are distributed across all outbound lanes by
target path. Per-path message ordering is preserved (same path → same
lane). PriorityMessage routing and all other message types are unaffected.
Tests:
- sbt "remote / Test / compile" — passes
- sbt "remote / Test / testOnly *ActorSelectionQueueDistributionSpec" — 5/5
- CI will exercise the artery variants
References:
Refs #3041, supersedes #3090
* test(artery): add unit tests for ActorSelection lane distribution
Verify that ActorSelection messages are distributed across outbound
lanes based on target path hash:
- Different target paths map to different lanes
- Queue indices are always non-negative (Integer.MIN_VALUE safe)
- Same target path always maps to the same lane (ordering preserved)
- Single lane config works correctly
- Paths don't all concentrate on lane 0 (regression test for original bug)
Tests: sbt "remote / Test / testOnly *ActorSelectionQueueDistributionSpec"
— 5/5
References: Refs #3092
* test(artery): reduce ActorSelection round-trips from 1000 to 100
The ActorSelection message path has inherent per-message overhead:
synchronous deliverSelection on the inbound thread (path resolution),
MessageContainerSerializer (larger payload), and path element encoding.
With 1-lane configs (especially TLS-TCP), 1000 round-trips × 4 senders
creates sustained pressure on the single inbound thread pipeline,
causing the test to exceed the 30s timeout on CI.
100 round-trips still validates ordering and concurrency while being
proportional to the overhead difference vs the ActorRef variant.
Tests: sbt "remote / Test / compile" — passes
References: Refs #3092
* fix(artery): distribute inbound actor selection messages
Motivation: PR #3092 distributes ActorSelection messages across outbound
lanes, but inbound lane selection still used the wire recipient. For
ActorSelection that recipient is typically the root guardian, so messages from
one origin still concentrate on one inbound lane.
Modification: Partition inbound ActorSelection envelopes by the selected
target path encoded in the SelectionEnvelope while preserving same target path
and origin ordering. Add deterministic outbound and inbound lane distribution
coverage. RemoteSendConsistencySpec is intentionally unchanged.
Result: ActorSelection traffic can use multiple inbound lanes without
weakening same-target ordering or changing the wire protocol.
Tests: sbt "remote / Test / testOnly
org.apache.pekko.remote.artery.ActorSelectionQueueDistributionSpec"; sbt
"remote / Test / testOnly
org.apache.pekko.remote.artery.ArteryUpdSendConsistencyWithOneLaneSpec
org.apache.pekko.remote.artery.ArteryUpdSendConsistencyWithThreeLanesSpec
org.apache.pekko.remote.artery.ArteryTcpSendConsistencyWithOneLaneSpec
org.apache.pekko.remote.artery.ArteryTcpSendConsistencyWithThreeLanesSpec
org.apache.pekko.remote.artery.ArteryTlsTcpSendConsistencyWithO [...]
References: Refs #3092
* test(artery): keep send consistency spec unchanged
Motivation: The ActorSelection lane distribution fix should not rely on
changing RemoteSendConsistencySpec.
Modification: Restore RemoteSendConsistencySpec to the main branch version
so it is removed from the PR diff.
Result: The PR keeps the existing send consistency spec unchanged while
retaining the inbound ActorSelection lane distribution fix and focused
distribution coverage.
Tests: git diff --check; focused tests were run before this restore: sbt
"remote / Test / testOnly
org.apache.pekko.remote.artery.ActorSelectionQueueDistributionSpec" and sbt
"remote / Test / testOnly
org.apache.pekko.remote.artery.ArteryUpdSendConsistencyWithOneLaneSpec
org.apache.pekko.remote.artery.ArteryUpdSendConsistencyWithThreeLanesSpec
org.apache.pekko.remote.artery.ArteryTcpSendConsistencyWithOneLaneSpec
org.apache.pekko.remote.artery.ArteryTcpSendConsistencyWithThreeLanesSpe [...]
References: Refs #3092
* fix(artery): avoid full actor selection parse for lane hash
Motivation:
Inbound ActorSelection lane partitioning needs the selected target path
hash, but parsing the full SelectionEnvelope in the inbound hot path adds
avoidable allocation and CPU cost.
Modification:
Scan only the SelectionEnvelope pattern fields with CodedInputStream when
computing the lane hash. Skip unknown fields and fall back to the existing
recipient uid hash if scanning fails. Add coverage for unknown fields so the
read-side hash remains tolerant of rolling-upgrade wire data.
Result:
ActorSelection lane distribution keeps the same wire format and avoids full
protobuf object construction before normal deserialization.
Tests:
- sbt 'remote / Test / testOnly
org.apache.pekko.remote.artery.ActorSelectionQueueDistributionSpec' (passed, 11
tests)
- git diff --check (passed)
References:
Refs #3092
* test(artery): add SelectParent and SelectChildPattern coverage for
ActorSelection lane distribution
Motivation:
The existing ActorSelectionQueueDistributionSpec only tested
SelectChildName elements. SelectParent (type=0, no matcher) and
SelectChildPattern (type=2, with matcher) were untested, leaving gaps
in the CodedInputStream-based protobuf parser coverage.
Modification:
Add 4 new test cases verifying hash consistency between ByteBuffer and
SelectionEnvelope parsing for SelectParent and SelectChildPattern, and
that different SelectionPathElement types produce distinct hashes.
Result:
15/15 tests pass, covering all three SelectionPathElement variants.
Tests:
sbt "remote / Test / testOnly
*ActorSelectionQueueDistributionSpec" — 15/15 passed
References:
Refs #3092
---
.../pekko/remote/artery/ArteryTransport.scala | 101 +++++++-
.../apache/pekko/remote/artery/Association.scala | 22 +-
.../ActorSelectionQueueDistributionSpec.scala | 288 +++++++++++++++++++++
3 files changed, 403 insertions(+), 8 deletions(-)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
index ba3f8e650b..f88e236435 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
@@ -13,6 +13,7 @@
package org.apache.pekko.remote.artery
+import java.nio.ByteBuffer
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong
@@ -25,7 +26,7 @@ import scala.concurrent.Future
import scala.concurrent.Promise
import scala.concurrent.duration._
import scala.util.Try
-import scala.util.control.NoStackTrace
+import scala.util.control.{ NoStackTrace, NonFatal }
import org.apache.pekko
import pekko.Done
@@ -36,6 +37,8 @@ import pekko.dispatch.Dispatchers
import pekko.event.Logging
import pekko.event.MarkerLoggingAdapter
import pekko.remote.AddressUidExtension
+import pekko.remote.ContainerFormats
+import pekko.protobufv3.internal.CodedInputStream
import pekko.remote.RemoteActorRef
import pekko.remote.RemoteActorRefProvider
import pekko.remote.RemoteTransport
@@ -54,6 +57,7 @@ import pekko.stream._
import pekko.stream.scaladsl.Flow
import pekko.stream.scaladsl.Keep
import pekko.stream.scaladsl.Sink
+import pekko.serialization.SerializationExtension
import pekko.util.OptionVal
import pekko.util.WildcardIndex
@@ -332,6 +336,8 @@ private[remote] abstract class ArteryTransport(_system:
ExtendedActorSystem, _pr
private val testState = new SharedTestState
protected val inboundLanes = settings.Advanced.InboundLanes
+ private lazy val actorSelectionMessageSerializerId =
+
SerializationExtension(system).serializerFor(classOf[ActorSelectionMessage]).identifier
val largeMessageChannelEnabled: Boolean =
!settings.LargeMessageDestinations.wildcardTree.isEmpty ||
@@ -460,14 +466,16 @@ private[remote] abstract class ArteryTransport(_system:
ExtendedActorSystem, _pr
// Select inbound lane based on destination to preserve message order,
// Also include the uid of the sending system in the hash to spread
- // "hot" destinations, e.g. ActorSelection anchor.
+ // "hot" destinations. For ActorSelection, the recipient is typically the
+ // root guardian, so use the selected target path as the destination key.
protected val inboundLanePartitioner: InboundEnvelope => Int = env => {
env.recipient match {
case OptionVal.Some(r) =>
- val a = r.path.uid
- val b = env.originUid
- val hashA = 23 + a
- val hash: Int = 23 * hashA + java.lang.Long.hashCode(b)
+ val destinationHash =
+ if (env.serializer == actorSelectionMessageSerializerId &&
(env.envelopeBuffer ne null))
+
ArteryTransport.actorSelectionMessagePathHash(env.envelopeBuffer.byteBuffer).getOrElse(r.path.uid)
+ else r.path.uid
+ val hash = ArteryTransport.inboundLanePartitionHash(destinationHash,
env.originUid)
math.abs(hash % inboundLanes)
case _ =>
// the lane is set by the DuplicateHandshakeReq stage, otherwise 0
@@ -979,6 +987,87 @@ private[remote] object ArteryTransport {
val OrdinaryStreamId = 2
val LargeStreamId = 3
+ def inboundLanePartitionHash(destinationHash: Int, originUid: Long): Int = {
+ val hashA = 23 + destinationHash
+ 23 * hashA + java.lang.Long.hashCode(originUid)
+ }
+
+ def actorSelectionMessagePathHash(byteBuffer: ByteBuffer): Option[Int] = {
+ val copy = byteBuffer.asReadOnlyBuffer()
+ try Some(actorSelectionMessagePathHash(CodedInputStream.newInstance(copy)))
+ catch {
+ case NonFatal(_) => None
+ }
+ }
+
+ private val SelectionEnvelopePatternTag = 26
+ private val SelectionTypeTag = 8
+ private val SelectionMatcherTag = 18
+
+ private def actorSelectionMessagePathHash(input: CodedInputStream): Int = {
+ var hash = 23
+ var done = false
+ while (!done) {
+ input.readTag() match {
+ case 0 =>
+ done = true
+ case SelectionEnvelopePatternTag =>
+ hash = actorSelectionMessagePathHash(input, input.readRawVarint32(),
hash)
+ case tag =>
+ if (!input.skipField(tag))
+ done = true
+ }
+ }
+ hash
+ }
+
+ private def actorSelectionMessagePathHash(input: CodedInputStream, length:
Int, currentHash: Int): Int = {
+ val oldLimit = input.pushLimit(length)
+ var patternType = -1
+ var matcherHash: Option[Int] = None
+ var done = false
+ try {
+ while (!done) {
+ input.readTag() match {
+ case 0 =>
+ done = true
+ case SelectionTypeTag =>
+ patternType = input.readEnum()
+ case SelectionMatcherTag =>
+ matcherHash = Some(input.readStringRequireUtf8().hashCode)
+ case tag =>
+ if (!input.skipField(tag))
+ done = true
+ }
+ }
+ } finally {
+ input.popLimit(oldLimit)
+ }
+
+ if (patternType == -1)
+ throw new IllegalArgumentException("ActorSelection Selection pattern
without type")
+
+ val hash = 23 * currentHash + patternType
+ matcherHash match {
+ case Some(matcher) => 23 * hash + matcher
+ case None => hash
+ }
+ }
+
+ def actorSelectionMessagePathHash(selectionEnvelope:
ContainerFormats.SelectionEnvelope): Int = {
+ var hash = 23
+ val patterns = selectionEnvelope.getPatternList
+ var i = 0
+ while (i < patterns.size) {
+ val pattern = patterns.get(i)
+ hash = 23 * hash + pattern.getType.getNumber
+ if (pattern.hasMatcher)
+ hash = 23 * hash + pattern.getMatcher.hashCode
+ i += 1
+ }
+ hash
+ }
+
def streamName(streamId: Int): String =
streamId match {
case ControlStreamId => "control"
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
index eb9bf1176a..f0fc935036 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
@@ -424,8 +424,26 @@ private[remote] class Association(
}(materializer.executionContext)
case _: SystemMessage =>
sendSystemMessage(outboundEnvelope)
- case ActorSelectionMessage(_: PriorityMessage, _, _) | _:
ControlMessage | _: ClearSystemMessageDelivery =>
- // ActorSelectionMessage with PriorityMessage is used by cluster
and remote failure detector heartbeating
+ case sel: ActorSelectionMessage =>
+ sel.msg match {
+ case _: PriorityMessage =>
+ // ActorSelectionMessage with PriorityMessage is used by
cluster and remote failure detector heartbeating
+ if (!controlQueue.offer(outboundEnvelope)) {
+ dropped(ControlQueueIndex, controlQueueSize,
outboundEnvelope)
+ }
+ case _ =>
+ // Distribute ActorSelection messages across lanes based on
the target
+ // path hash rather than the anchor's (typically root
guardian, UID=0).
+ // Without this, all ActorSelection messages concentrate on a
single
+ // outbound lane, creating a throughput bottleneck.
+ val queueIndex =
+ if (outboundLanes == 1) OrdinaryQueueIndex
+ else OrdinaryQueueIndex + ((sel.elements.hashCode() &
Int.MaxValue) % outboundLanes)
+ val queue = queues(queueIndex)
+ if (!queue.offer(outboundEnvelope))
+ dropped(queueIndex, queueSize, outboundEnvelope)
+ }
+ case _: ControlMessage | _: ClearSystemMessageDelivery =>
if (!controlQueue.offer(outboundEnvelope)) {
dropped(ControlQueueIndex, controlQueueSize, outboundEnvelope)
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/ActorSelectionQueueDistributionSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/ActorSelectionQueueDistributionSpec.scala
new file mode 100644
index 0000000000..a876940506
--- /dev/null
+++
b/remote/src/test/scala/org/apache/pekko/remote/artery/ActorSelectionQueueDistributionSpec.scala
@@ -0,0 +1,288 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote.artery
+
+import java.nio.ByteBuffer
+
+import org.apache.pekko
+import pekko.actor.SelectChildName
+import pekko.actor.SelectChildPattern
+import pekko.actor.SelectParent
+import pekko.actor.SelectionPathElement
+import pekko.protobufv3.internal.ByteString
+import pekko.protobufv3.internal.UnknownFieldSet
+import pekko.remote.ContainerFormats
+import pekko.remote.ContainerFormats.PatternType
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class ActorSelectionQueueDistributionSpec extends AnyWordSpec with Matchers {
+
+ private val OrdinaryQueueIndex = 2
+
+ private def computeOutboundQueueIndex(elements: Seq[SelectionPathElement],
outboundLanes: Int): Int = {
+ if (outboundLanes == 1) OrdinaryQueueIndex
+ else OrdinaryQueueIndex + ((elements.hashCode() & Int.MaxValue) %
outboundLanes)
+ }
+
+ private def selectionElements(names: String*): Seq[SelectionPathElement] =
+ names.map(SelectChildName.apply)
+
+ private def selectionEnvelope(names: String*):
ContainerFormats.SelectionEnvelope = {
+ val builder = ContainerFormats.SelectionEnvelope
+ .newBuilder()
+ .setEnclosedMessage(ByteString.EMPTY)
+ .setSerializerId(0)
+
+ names.foreach { name =>
+ builder.addPattern(
+ ContainerFormats.Selection
+ .newBuilder()
+ .setType(PatternType.CHILD_NAME)
+ .setMatcher(name))
+ }
+
+ builder.build()
+ }
+
+ private def selectionEnvelopeFromElements(elements: SelectionPathElement*):
ContainerFormats.SelectionEnvelope = {
+ val builder = ContainerFormats.SelectionEnvelope
+ .newBuilder()
+ .setEnclosedMessage(ByteString.EMPTY)
+ .setSerializerId(0)
+
+ elements.foreach {
+ case SelectChildName(name) =>
+ builder.addPattern(
+ ContainerFormats.Selection
+ .newBuilder()
+ .setType(PatternType.CHILD_NAME)
+ .setMatcher(name))
+ case SelectChildPattern(patternStr) =>
+ builder.addPattern(
+ ContainerFormats.Selection
+ .newBuilder()
+ .setType(PatternType.CHILD_PATTERN)
+ .setMatcher(patternStr))
+ case SelectParent =>
+ builder.addPattern(
+ ContainerFormats.Selection
+ .newBuilder()
+ .setType(PatternType.PARENT))
+ }
+
+ builder.build()
+ }
+
+ private def unknownField(fieldNumber: Int): UnknownFieldSet =
+ UnknownFieldSet
+ .newBuilder()
+ .addField(fieldNumber,
UnknownFieldSet.Field.newBuilder().addVarint(17L).build())
+ .build()
+
+ private def selectionEnvelopeWithUnknownFields(names: String*):
ContainerFormats.SelectionEnvelope = {
+ val builder = ContainerFormats.SelectionEnvelope
+ .newBuilder()
+ .setEnclosedMessage(ByteString.EMPTY)
+ .setSerializerId(0)
+ .setUnknownFields(unknownField(42))
+
+ names.foreach { name =>
+ builder.addPattern(
+ ContainerFormats.Selection
+ .newBuilder()
+ .setType(PatternType.CHILD_NAME)
+ .setMatcher(name)
+ .setUnknownFields(unknownField(43)))
+ }
+
+ builder.build()
+ }
+
+ private def computeInboundLane(names: Seq[String], inboundLanes: Int,
originUid: Long): Int = {
+ val selectionHash =
ArteryTransport.actorSelectionMessagePathHash(selectionEnvelope(names: _*))
+ math.abs(ArteryTransport.inboundLanePartitionHash(selectionHash,
originUid) % inboundLanes)
+ }
+
+ "ActorSelection queue distribution" must {
+
+ "distribute different outbound target paths across lanes" in {
+ val outboundLanes = 3
+ val paths = (1 to 100).map(i => selectionElements("user", s"echo$i"))
+ val queueIndices = paths.map(p => computeOutboundQueueIndex(p,
outboundLanes))
+ val distinctLanes = queueIndices.distinct
+
+ distinctLanes.size should be > 1
+ distinctLanes.foreach { idx =>
+ idx should be >= OrdinaryQueueIndex
+ idx should be < (OrdinaryQueueIndex + outboundLanes)
+ }
+ }
+
+ "produce non-negative queue indices even for negative hash codes" in {
+ val outboundLanes = 3
+ val paths = (1 to 1000).map(i => selectionElements("user", s"actor$i"))
+ val queueIndices = paths.map(p => computeOutboundQueueIndex(p,
outboundLanes))
+
+ queueIndices.foreach { idx =>
+ idx should be >= OrdinaryQueueIndex
+ }
+ }
+
+ "preserve ordering for same target path" in {
+ val outboundLanes = 3
+ val path = selectionElements("user", "echoA")
+ val indices = (1 to 10).map(_ => computeOutboundQueueIndex(path,
outboundLanes))
+
+ indices.distinct.size should be(1)
+ }
+
+ "use single lane when outboundLanes is 1" in {
+ val paths = (1 to 10).map(i => selectionElements("user", s"echo$i"))
+ val queueIndices = paths.map(p => computeOutboundQueueIndex(p, 1))
+
+ queueIndices.distinct should be(Seq(OrdinaryQueueIndex))
+ }
+
+ "not concentrate all paths on lane 0 (original bug)" in {
+ val outboundLanes = 3
+ val paths = Seq(
+ selectionElements("user", "echoA2"),
+ selectionElements("user", "echoB2"),
+ selectionElements("user", "echoC2"))
+
+ val queueIndices = paths.map(p => computeOutboundQueueIndex(p,
outboundLanes))
+ val laneOffsets = queueIndices.map(_ - OrdinaryQueueIndex)
+
+ laneOffsets should not be Seq(0, 0, 0)
+ }
+
+ "distribute inbound ActorSelection messages by selected target path" in {
+ val inboundLanes = 3
+ val originUid = 17L
+ val paths = (1 to 100).map(i => Seq("user", s"echo$i"))
+ val lanes = paths.map(path => computeInboundLane(path, inboundLanes,
originUid))
+
+ lanes.distinct.size should be > 1
+ }
+
+ "not concentrate inbound ActorSelection messages on the root guardian
lane" in {
+ val inboundLanes = 3
+ val originUid = 17L
+ val paths = Seq(
+ Seq("user", "echoA2"),
+ Seq("user", "echoB2"),
+ Seq("user", "echoC2"))
+ val lanes = paths.map(path => computeInboundLane(path, inboundLanes,
originUid))
+ val rootGuardianLane =
+ math.abs(ArteryTransport.inboundLanePartitionHash(destinationHash = 0,
originUid) % inboundLanes)
+
+ lanes should not be paths.map(_ => rootGuardianLane)
+ }
+
+ "preserve inbound ordering for same selected target path and origin" in {
+ val inboundLanes = 3
+ val originUid = 17L
+ val path = Seq("user", "echoA2")
+ val lanes = (1 to 10).map(_ => computeInboundLane(path, inboundLanes,
originUid))
+
+ lanes.distinct.size should be(1)
+ }
+
+ "parse ActorSelection path hash without moving the envelope buffer
position" in {
+ val envelope = selectionEnvelope("user", "echoA2")
+ val bytes = envelope.toByteArray
+ val byteBuffer = ByteBuffer.allocate(bytes.length + 4)
+ byteBuffer.putInt(17)
+ byteBuffer.put(bytes)
+ byteBuffer.flip()
+ byteBuffer.position(4)
+
+ val positionBefore = byteBuffer.position()
+ ArteryTransport.actorSelectionMessagePathHash(byteBuffer) shouldBe Some(
+ ArteryTransport.actorSelectionMessagePathHash(envelope))
+
+ byteBuffer.position() shouldBe positionBefore
+ }
+
+ "skip unknown fields when parsing ActorSelection path hash" in {
+ val envelope = selectionEnvelopeWithUnknownFields("user", "echoA2")
+ val byteBuffer = ByteBuffer.wrap(envelope.toByteArray)
+
+ ArteryTransport.actorSelectionMessagePathHash(byteBuffer) shouldBe Some(
+ ArteryTransport.actorSelectionMessagePathHash(envelope))
+ }
+
+ "ignore invalid ActorSelection envelopes when computing path hash" in {
+ val byteBuffer = ByteBuffer.wrap(Array[Byte](1, 2, 3))
+
+ ArteryTransport.actorSelectionMessagePathHash(byteBuffer) shouldBe None
+ }
+
+ "produce consistent hash for SelectParent elements between ByteBuffer and
SelectionEnvelope parsing" in {
+ val envelope = selectionEnvelopeFromElements(SelectChildName("user"),
SelectParent, SelectChildName("echo"))
+ val byteBuffer = ByteBuffer.wrap(envelope.toByteArray)
+
+ ArteryTransport.actorSelectionMessagePathHash(byteBuffer) shouldBe Some(
+ ArteryTransport.actorSelectionMessagePathHash(envelope))
+ }
+
+ "produce consistent hash for SelectChildPattern elements between
ByteBuffer and SelectionEnvelope parsing" in {
+ val envelope =
+ selectionEnvelopeFromElements(SelectChildName("user"),
SelectChildPattern("echo*"))
+ val byteBuffer = ByteBuffer.wrap(envelope.toByteArray)
+
+ ArteryTransport.actorSelectionMessagePathHash(byteBuffer) shouldBe Some(
+ ArteryTransport.actorSelectionMessagePathHash(envelope))
+ }
+
+ "produce distinct hashes for different SelectionPathElement types" in {
+ val childNameEnvelope =
selectionEnvelopeFromElements(SelectChildName("user"), SelectChildName("echo"))
+ val selectParentEnvelope =
+ selectionEnvelopeFromElements(SelectChildName("user"), SelectParent,
SelectChildName("echo"))
+ val childPatternEnvelope =
+ selectionEnvelopeFromElements(SelectChildName("user"),
SelectChildPattern("echo*"))
+
+ val childNameHash =
ArteryTransport.actorSelectionMessagePathHash(childNameEnvelope)
+ val selectParentHash =
ArteryTransport.actorSelectionMessagePathHash(selectParentEnvelope)
+ val childPatternHash =
ArteryTransport.actorSelectionMessagePathHash(childPatternEnvelope)
+
+ childNameHash should not be selectParentHash
+ childNameHash should not be childPatternHash
+ selectParentHash should not be childPatternHash
+ }
+
+ "distribute paths with SelectParent across inbound lanes" in {
+ val inboundLanes = 3
+ val originUid = 42L
+ val elements = Seq(
+ Seq[SelectionPathElement](SelectChildName("user"), SelectParent,
SelectChildName("echoA")),
+ Seq[SelectionPathElement](SelectChildName("user"), SelectParent,
SelectChildName("echoB")),
+ Seq[SelectionPathElement](SelectChildName("user"), SelectParent,
SelectChildName("echoC")))
+
+ val lanes = elements.map { elems =>
+ val envelope = selectionEnvelopeFromElements(elems: _*)
+ val selectionHash =
ArteryTransport.actorSelectionMessagePathHash(envelope)
+ math.abs(ArteryTransport.inboundLanePartitionHash(selectionHash,
originUid) % inboundLanes)
+ }
+
+ lanes.distinct.size should be > 1
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
