raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Optimized serializer for ORSet[ActorRef], #23703

Browse source 
* ORSet[ActorRef] is used by Akka Typed receptionist
* and similar for GSet[ActorRef]

ORSetSerializationBenchmark
This commit is contained in:
Patrik Nordwall 2018-06-05 14:15:46 +02:00
parent a05170c419
commit f754705c9c
7 changed files with 742 additions and 84 deletions
Download patch file Download diff file Expand all files Collapse all files

57
akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatedDataSerializer.scala
View file

@ -13,6 +13,7 @@ import java.util.TreeSet
import scala.annotation.tailrec
import scala.collection.JavaConverters._
import scala.collection.breakOut
import akka.actor.ExtendedActorSystem
import akka.cluster.ddata._
import akka.cluster.ddata.Replicator.Internal._
@ -22,10 +23,11 @@ import akka.serialization.SerializerWithStringManifest
import akka.serialization.BaseSerializer
import akka.protobuf.{ ByteString, GeneratedMessage }
import akka.util.ByteString.UTF_8
import java.io.NotSerializableException
import akka.actor.ActorRef
import akka.cluster.ddata.protobuf.msg.ReplicatorMessages.OtherMessage
import akka.serialization.Serialization
private object ReplicatedDataSerializer {
/*
@ -321,11 +323,13 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
val intElements = new ArrayList[Integer]
val longElements = new ArrayList[jl.Long]
val otherElements = new ArrayList[dm.OtherMessage]
val actorRefElements = new ArrayList[String]
gset.elements.foreach {
case s: String ⇒ stringElements.add(s)
case i: Int ⇒ intElements.add(i)
case l: Long ⇒ longElements.add(l)
case other ⇒ otherElements.add(otherMessageToProto(other))
case s: String ⇒ stringElements.add(s)
case i: Int ⇒ intElements.add(i)
case l: Long ⇒ longElements.add(l)
case ref: ActorRef ⇒ actorRefElements.add(Serialization.serializedActorPath(ref))
case other ⇒ otherElements.add(otherMessageToProto(other))
}
if (!stringElements.isEmpty) {
Collections.sort(stringElements)
@ -343,17 +347,26 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
Collections.sort(otherElements, OtherMessageComparator)
b.addAllOtherElements(otherElements)
}
if (!actorRefElements.isEmpty) {
Collections.sort(actorRefElements)
b.addAllActorRefElements(actorRefElements)
}
b.build()
}
def gsetFromBinary(bytes: Array[Byte]): GSet[_] =
gsetFromProto(rd.GSet.parseFrom(bytes))
def gsetFromProto(gset: rd.GSet): GSet[Any] =
GSet(gset.getStringElementsList.iterator.asScala.toSet ++
gset.getIntElementsList.iterator.asScala ++
gset.getLongElementsList.iterator.asScala ++
gset.getOtherElementsList.iterator.asScala.map(otherMessageFromProto))
def gsetFromProto(gset: rd.GSet): GSet[Any] = {
val elements: Iterator[Any] = {
gset.getStringElementsList.iterator.asScala ++
gset.getIntElementsList.iterator.asScala ++
gset.getLongElementsList.iterator.asScala ++
gset.getOtherElementsList.iterator.asScala.map(otherMessageFromProto) ++
gset.getActorRefElementsList.iterator.asScala.map(resolveActorRef)
}
GSet(elements.toSet)
}
def orsetToProto(orset: ORSet[_]): rd.ORSet =
orsetToProtoImpl(orset.asInstanceOf[ORSet[Any]])
@ -366,10 +379,12 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
val longElements = new ArrayList[jl.Long]
val otherElements = new ArrayList[dm.OtherMessage]
var otherElementsMap = Map.empty[dm.OtherMessage, Any]
val actorRefElements = new ArrayList[ActorRef]
orset.elementsMap.keysIterator.foreach {
case s: String ⇒ stringElements.add(s)
case i: Int ⇒ intElements.add(i)
case l: Long ⇒ longElements.add(l)
case s: String ⇒ stringElements.add(s)
case i: Int ⇒ intElements.add(i)
case l: Long ⇒ longElements.add(l)
case ref: ActorRef ⇒ actorRefElements.add(ref)
case other ⇒
val enclosedMsg = otherMessageToProto(other)
otherElements.add(enclosedMsg)
@ -409,6 +424,14 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
b.addAllOtherElements(otherElements)
addDots(otherElements)
}
if (!actorRefElements.isEmpty) {
Collections.sort(actorRefElements)
val iter = actorRefElements.iterator
while (iter.hasNext) {
b.addActorRefElements(Serialization.serializedActorPath(iter.next()))
}
addDots(actorRefElements)
}
b.build()
}
@ -463,11 +486,13 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
}
def orsetFromProto(orset: rd.ORSet): ORSet[Any] = {
val elements: Iterator[Any] =
(orset.getStringElementsList.iterator.asScala ++
val elements: Iterator[Any] = {
orset.getStringElementsList.iterator.asScala ++
orset.getIntElementsList.iterator.asScala ++
orset.getLongElementsList.iterator.asScala ++
orset.getOtherElementsList.iterator.asScala.map(otherMessageFromProto))
orset.getOtherElementsList.iterator.asScala.map(otherMessageFromProto) ++
orset.getActorRefElementsList.iterator.asScala.map(resolveActorRef)
}
val dots = orset.getDotsList.asScala.map(versionVectorFromProto).iterator
val elementsMap = elements.zip(dots).toMap