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

Refactoring of outbound compression, #21210

Browse source 
* outbound compression is now immutable, by simply using
  CompressionTable[ActorRef] and CompressionTable[String]
* immutable outbound compression will make it possible to use
  them from multiple Encoder instances, when we add several lanes
  for parallel serialization
* outbound compression tables not shared via AssociationState
* the advertised tables are sent to the Encoder stage via async
  callback, no need to reference the tables in other places than
  the Encoder stage, no more races via shared mutable state
* when outbound stream is started or restarted it can start out
  without compression, until next advertisement is received
* ensure outbound compression is cleared before handshake is signaled complete
This commit is contained in:
Patrik Nordwall 2016-08-24 19:52:07 +02:00
parent af5eb4c6bf
commit 0c0e3c5efd
16 changed files with 454 additions and 540 deletions
Download patch file Download diff file Expand all files Collapse all files

13
akka-bench-jmh/src/main/scala/akka/remote/artery/CodecBenchmark.scala
View file

@ -29,6 +29,8 @@ import com.typesafe.config.ConfigFactory
import org.openjdk.jmh.annotations._
import akka.util.OptionVal
import akka.actor.Address
import scala.concurrent.Future
import akka.Done
@State(Scope.Benchmark)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@ -57,7 +59,6 @@ class CodecBenchmark {
private val inboundEnvelopePool = ReusableInboundEnvelope.createObjectPool(capacity = 16)
private val outboundEnvelopePool = ReusableOutboundEnvelope.createObjectPool(capacity = 16)
val compressionOut = NoOutboundCompressions
val headerIn = HeaderBuilder.in(NoInboundCompressions)
val envelopeTemplateBuffer = ByteBuffer.allocate(ArteryTransport.MaximumFrameSize).order(ByteOrder.LITTLE_ENDIAN)
@ -73,7 +74,7 @@ class CodecBenchmark {
// the following methods are not used by in this test
override def sendControl(to: Address, message: ControlMessage): Unit = ???
override def association(remoteAddress: Address): OutboundContext = ???
override def completeHandshake(peer: UniqueAddress): Unit = ???
override def completeHandshake(peer: UniqueAddress): Future[Done] = ???
}
private var materializer: ActorMaterializer = _
@ -136,8 +137,8 @@ class CodecBenchmark {
val latch = new CountDownLatch(1)
val N = 100000
val encoder: Flow[OutboundEnvelope, EnvelopeBuffer, NotUsed] =
Flow.fromGraph(new Encoder(uniqueLocalAddress, system, compressionOut, outboundEnvelopePool, envelopePool))
val encoder: Flow[OutboundEnvelope, EnvelopeBuffer, Encoder.ChangeOutboundCompression] =
Flow.fromGraph(new Encoder(uniqueLocalAddress, system, outboundEnvelopePool, envelopePool))
Source.fromGraph(new BenchTestSourceSameElement(N, "elem"))
.map(msg ⇒ outboundEnvelopePool.acquire().init(OptionVal.None, payload, OptionVal.Some(remoteRefB)))
@ -193,8 +194,8 @@ class CodecBenchmark {
val latch = new CountDownLatch(1)
val N = 100000
val encoder: Flow[OutboundEnvelope, EnvelopeBuffer, NotUsed] =
Flow.fromGraph(new Encoder(uniqueLocalAddress, system, compressionOut, outboundEnvelopePool, envelopePool))
val encoder: Flow[OutboundEnvelope, EnvelopeBuffer, Encoder.ChangeOutboundCompression] =
Flow.fromGraph(new Encoder(uniqueLocalAddress, system, outboundEnvelopePool, envelopePool))
val localRecipient = resolvedRef.path.toSerializationFormatWithAddress(uniqueLocalAddress.address)
val provider = RARP(system).provider

2
akka-bench-jmh/src/main/scala/akka/remote/compress/HeavyHittersBenchmark.scala
View file

@ -49,7 +49,7 @@ class HeavyHittersBenchmark {
@Param(Array("8192"))
var n: Int = 0
var topN: TopHeavyHitters[String] = _
private var topN: TopHeavyHitters[String] = _
val rand = new Random(1001021)

125
akka-remote/src/main/scala/akka/remote/artery/ArteryTransport.scala
View file

@ -3,22 +3,30 @@
*/
package akka.remote.artery
import java.io.File
import java.net.InetSocketAddress
import java.nio.channels.{ DatagramChannel, FileChannel }
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.TimeUnit.MICROSECONDS
import akka.remote.artery.compress.CompressionProtocol.CompressionMessage
import scala.annotation.tailrec
import scala.collection.JavaConverters._
import scala.concurrent.Future
import scala.concurrent.Promise
import scala.concurrent.duration._
import scala.util.Failure
import scala.util.Success
import scala.util.Try
import scala.util.control.NonFatal
import akka.Done
import akka.NotUsed
import akka.actor._
import akka.actor.Actor
import akka.actor.Cancellable
import akka.actor.Props
import akka.event.Logging
import akka.event.LoggingAdapter
import akka.remote.AddressUidExtension
@ -30,13 +38,17 @@ import akka.remote.RemoteTransport
import akka.remote.RemotingLifecycleEvent
import akka.remote.ThisActorSystemQuarantinedEvent
import akka.remote.UniqueAddress
import akka.remote.artery.Encoder.ChangeOutboundCompression
import akka.remote.artery.InboundControlJunction.ControlMessageObserver
import akka.remote.artery.InboundControlJunction.ControlMessageSubject
import akka.remote.artery.OutboundControlJunction.OutboundControlIngress
import akka.remote.artery.compress._
import akka.remote.artery.compress.CompressionProtocol.CompressionMessage
import akka.remote.transport.AkkaPduCodec
import akka.remote.transport.AkkaPduProtobufCodec
import akka.remote.artery.compress._
import akka.stream.AbruptTerminationException
import akka.stream.ActorMaterializer
import akka.stream.ActorMaterializerSettings
import akka.stream.KillSwitches
import akka.stream.Materializer
import akka.stream.SharedKillSwitch
@ -46,35 +58,19 @@ import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Source
import akka.util.Helpers.ConfigOps
import akka.util.Helpers.Requiring
import akka.util.OptionVal
import akka.util.WildcardIndex
import io.aeron.Aeron
import io.aeron.AvailableImageHandler
import io.aeron.CncFileDescriptor
import io.aeron.Image
import io.aeron.UnavailableImageHandler
import io.aeron.driver.MediaDriver
import io.aeron.driver.ThreadingMode
import io.aeron.exceptions.ConductorServiceTimeoutException
import org.agrona.ErrorHandler
import org.agrona.IoUtil
import java.io.File
import java.net.InetSocketAddress
import java.nio.channels.{ DatagramChannel, FileChannel }
import akka.remote.artery.OutboundControlJunction.OutboundControlIngress
import io.aeron.CncFileDescriptor
import java.util.concurrent.atomic.AtomicLong
import scala.collection.JavaConverters._
import akka.stream.ActorMaterializerSettings
import scala.annotation.tailrec
import akka.util.OptionVal
import io.aeron.driver.ThreadingMode
import org.agrona.concurrent.BackoffIdleStrategy
import org.agrona.concurrent.BusySpinIdleStrategy
import scala.util.control.NonFatal
import akka.actor.Props
import akka.actor.Actor
/**
* INTERNAL API
@ -105,7 +101,7 @@ private[akka] trait InboundContext {
*/
def association(uid: Long): OptionVal[OutboundContext]
def completeHandshake(peer: UniqueAddress): Unit
def completeHandshake(peer: UniqueAddress): Future[Done]
}
@ -117,8 +113,7 @@ private[akka] object AssociationState {
new AssociationState(
incarnation = 1,
uniqueRemoteAddressPromise = Promise(),
quarantined = ImmutableLongMap.empty[QuarantinedTimestamp],
outboundCompressions = NoOutboundCompressions)
quarantined = ImmutableLongMap.empty[QuarantinedTimestamp])
final case class QuarantinedTimestamp(nanoTime: Long) {
override def toString: String =
@ -130,10 +125,9 @@ private[akka] object AssociationState {
* INTERNAL API
*/
private[akka] final class AssociationState(
val incarnation: Int,
val incarnation: Int,
val uniqueRemoteAddressPromise: Promise[UniqueAddress],
val quarantined: ImmutableLongMap[AssociationState.QuarantinedTimestamp],
val outboundCompressions: OutboundCompressions) {
val quarantined: ImmutableLongMap[AssociationState.QuarantinedTimestamp]) {
import AssociationState.QuarantinedTimestamp
@ -159,11 +153,8 @@ private[akka] final class AssociationState(
}
}
def withCompression(compression: OutboundCompressions) =
new AssociationState(incarnation, uniqueRemoteAddressPromise, quarantined, compression)
def newIncarnation(remoteAddressPromise: Promise[UniqueAddress], compression: OutboundCompressions): AssociationState =
new AssociationState(incarnation + 1, remoteAddressPromise, quarantined, compression)
def newIncarnation(remoteAddressPromise: Promise[UniqueAddress]): AssociationState =
new AssociationState(incarnation + 1, remoteAddressPromise, quarantined)
def newQuarantined(): AssociationState =
uniqueRemoteAddressPromise.future.value match {
@ -171,8 +162,7 @@ private[akka] final class AssociationState(
new AssociationState(
incarnation,
uniqueRemoteAddressPromise,
quarantined = quarantined.updated(a.uid, QuarantinedTimestamp(System.nanoTime())),
outboundCompressions = NoOutboundCompressions) // after quarantine no compression needed anymore, drop it
quarantined = quarantined.updated(a.uid, QuarantinedTimestamp(System.nanoTime())))
case _ ⇒ this
}
@ -235,7 +225,7 @@ private[akka] trait OutboundContext {
*/
private[remote] object FlushOnShutdown {
def props(done: Promise[Done], timeout: FiniteDuration,
inboundContext: InboundContext, associations: Set[Association]): Props = {
inboundContext: InboundContext, associations: Set[Association]): Props = {
require(associations.nonEmpty)
Props(new FlushOnShutdown(done, timeout, inboundContext, associations))
}
@ -247,7 +237,7 @@ private[remote] object FlushOnShutdown {
* INTERNAL API
*/
private[remote] class FlushOnShutdown(done: Promise[Done], timeout: FiniteDuration,
inboundContext: InboundContext, associations: Set[Association]) extends Actor {
inboundContext: InboundContext, associations: Set[Association]) extends Actor {
var remaining = associations.flatMap(_.associationState.uniqueRemoteAddressValue)
@ -330,7 +320,8 @@ private[remote] class ArteryTransport(_system: ExtendedActorSystem, _provider: R
private val priorityMessageDestinations =
WildcardIndex[NotUsed]()
// this comes from remoting so is semi-ok to be hardcoded here
// These destinations are not defined in configuration because it should not
// be possible to abuse the control channel
.insert(Array("system", "remote-watcher"), NotUsed)
// these belongs to cluster and should come from there
.insert(Array("system", "cluster", "core", "daemon", "heartbeatSender"), NotUsed)
@ -579,17 +570,27 @@ private[remote] class ArteryTransport(_system: ExtendedActorSystem, _provider: R
case ActorRefCompressionAdvertisement(from, table) ⇒
log.debug("Incoming ActorRef compression advertisement from [{}], table: [{}]", from, table)
val a = association(from.address)
a.outboundCompression.applyActorRefCompressionTable(table)
a.sendControl(ActorRefCompressionAdvertisementAck(localAddress, table.version))
system.eventStream.publish(Events.ReceivedActorRefCompressionTable(from, table))
// make sure uid is same for active association
if (a.associationState.uniqueRemoteAddressValue().contains(from)) {
import system.dispatcher
a.changeActorRefCompression(table).foreach { _ ⇒
a.sendControl(ActorRefCompressionAdvertisementAck(localAddress, table.version))
system.eventStream.publish(Events.ReceivedActorRefCompressionTable(from, table))
}
}
case ActorRefCompressionAdvertisementAck(from, tableVersion) ⇒
inboundCompressions.foreach(_.confirmActorRefCompressionAdvertisement(from.uid, tableVersion))
case ClassManifestCompressionAdvertisement(from, table) ⇒
log.debug("Incoming Class Manifest compression advertisement from [{}], table: [{}]", from, table)
val a = association(from.address)
a.outboundCompression.applyClassManifestCompressionTable(table)
a.sendControl(ClassManifestCompressionAdvertisementAck(localAddress, table.version))
system.eventStream.publish(Events.ReceivedClassManifestCompressionTable(from, table))
// make sure uid is same for active association
if (a.associationState.uniqueRemoteAddressValue().contains(from)) {
import system.dispatcher
a.changeClassManifestCompression(table).foreach { _ ⇒
a.sendControl(ClassManifestCompressionAdvertisementAck(localAddress, table.version))
system.eventStream.publish(Events.ReceivedClassManifestCompressionTable(from, table))
}
}
case ClassManifestCompressionAdvertisementAck(from, tableVersion) ⇒
inboundCompressions.foreach(_.confirmActorRefCompressionAdvertisement(from.uid, tableVersion))
}
@ -719,8 +720,8 @@ private[remote] class ArteryTransport(_system: ExtendedActorSystem, _provider: R
if (testStages.isEmpty)
Future.successful(false)
else {
import scala.collection.JavaConverters._
import system.dispatcher
import scala.collection.JavaConverters._
val allTestStages = testStages.asScala.toVector ++ associationRegistry.allAssociations.flatMap(_.testStages)
Future.sequence(allTestStages.map(_.send(cmd))).map(_ ⇒ true)
}
@ -752,7 +753,7 @@ private[remote] class ArteryTransport(_system: ExtendedActorSystem, _provider: R
override def association(uid: Long): OptionVal[Association] =
associationRegistry.association(uid)
override def completeHandshake(peer: UniqueAddress): Unit = {
override def completeHandshake(peer: UniqueAddress): Future[Done] = {
val a = associationRegistry.setUID(peer)
a.completeHandshake(peer)
}
@ -765,19 +766,22 @@ private[remote] class ArteryTransport(_system: ExtendedActorSystem, _provider: R
association(remoteAddress).quarantine(reason = "", uid.map(_.toLong))
}
def outbound(outboundContext: OutboundContext, compression: OutboundCompressions): Sink[OutboundEnvelope, Future[Done]] =
createOutboundSink(ordinaryStreamId, outboundContext, compression, envelopePool)
def outbound(outboundContext: OutboundContext): Sink[OutboundEnvelope, (ChangeOutboundCompression, Future[Done])] =
createOutboundSink(ordinaryStreamId, outboundContext, envelopePool)
def outboundLarge(outboundContext: OutboundContext, compression: OutboundCompressions): Sink[OutboundEnvelope, Future[Done]] =
createOutboundSink(largeStreamId, outboundContext, compression, largeEnvelopePool)
def outboundLarge(outboundContext: OutboundContext): Sink[OutboundEnvelope, Future[Done]] =
createOutboundSink(largeStreamId, outboundContext, largeEnvelopePool)
.mapMaterializedValue { case (_, d) ⇒ d }
private def createOutboundSink(streamId: Int, outboundContext: OutboundContext,
bufferPool: EnvelopeBufferPool): Sink[OutboundEnvelope, (ChangeOutboundCompression, Future[Done])] = {
private def createOutboundSink(streamId: Int, outboundContext: OutboundContext, compression: OutboundCompressions, bufferPool: EnvelopeBufferPool): Sink[OutboundEnvelope, Future[Done]] = {
Flow.fromGraph(killSwitch.flow[OutboundEnvelope])
.via(new OutboundHandshake(system, outboundContext, outboundEnvelopePool, handshakeTimeout,
handshakeRetryInterval, injectHandshakeInterval))
.via(createEncoder(bufferPool, compression))
.viaMat(createEncoder(bufferPool))(Keep.right)
.toMat(new AeronSink(outboundChannel(outboundContext.remoteAddress), streamId, aeron, taskRunner,
envelopePool, giveUpSendAfter, createFlightRecorderEventSink()))(Keep.right)
envelopePool, giveUpSendAfter, createFlightRecorderEventSink()))(Keep.both)
}
/**
@ -794,25 +798,20 @@ private[remote] class ArteryTransport(_system: ExtendedActorSystem, _provider: R
// FIXME we can also add scrubbing stage that would collapse sys msg acks/nacks and remove duplicate Quarantine messages
}
def outboundControlPart2(outboundContext: OutboundContext, compression: OutboundCompressions): Sink[OutboundEnvelope, (OutboundControlIngress, Future[Done])] = {
def outboundControlPart2(outboundContext: OutboundContext): Sink[OutboundEnvelope, (OutboundControlIngress, Future[Done])] = {
Flow[OutboundEnvelope]
.viaMat(new OutboundControlJunction(outboundContext, outboundEnvelopePool))(Keep.right)
.via(encoder(compression))
.via(createEncoder(envelopePool))
.toMat(new AeronSink(outboundChannel(outboundContext.remoteAddress), controlStreamId, aeron, taskRunner,
envelopePool, Duration.Inf, createFlightRecorderEventSink()))(Keep.both)
}
def createEncoder(compression: OutboundCompressions, bufferPool: EnvelopeBufferPool): Flow[OutboundEnvelope, EnvelopeBuffer, NotUsed] =
Flow.fromGraph(new Encoder(localAddress, system, compression, outboundEnvelopePool, bufferPool))
private def createInboundCompressions(inboundContext: InboundContext): InboundCompressions =
if (remoteSettings.ArteryCompressionSettings.enabled) new InboundCompressionsImpl(system, inboundContext)
else NoInboundCompressions
def createEncoder(pool: EnvelopeBufferPool, compression: OutboundCompressions): Flow[OutboundEnvelope, EnvelopeBuffer, NotUsed] =
Flow.fromGraph(new Encoder(localAddress, system, compression, outboundEnvelopePool, pool))
def encoder(compression: OutboundCompressions): Flow[OutboundEnvelope, EnvelopeBuffer, NotUsed] = createEncoder(envelopePool, compression)
def createEncoder(pool: EnvelopeBufferPool): Flow[OutboundEnvelope, EnvelopeBuffer, ChangeOutboundCompression] =
Flow.fromGraph(new Encoder(localAddress, system, outboundEnvelopePool, pool))
def aeronSource(streamId: Int, pool: EnvelopeBufferPool): Source[EnvelopeBuffer, NotUsed] =
Source.fromGraph(new AeronSource(inboundChannel, streamId, aeron, taskRunner, pool,

178
akka-remote/src/main/scala/akka/remote/artery/Association.scala
View file

@ -4,41 +4,43 @@
package akka.remote.artery
import java.util.Queue
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicReference
import akka.remote.artery.compress.{ CompressionProtocol, CompressionTable, OutboundCompressions, OutboundCompressionsImpl }
import scala.annotation.tailrec
import scala.concurrent.Future
import scala.concurrent.Promise
import scala.concurrent.duration._
import scala.concurrent.duration.FiniteDuration
import scala.util.Success
import akka.{ Done, NotUsed }
import akka.actor.ActorRef
import akka.actor.ActorSelectionMessage
import akka.actor.Address
import akka.actor.RootActorPath
import akka.dispatch.sysmsg.SystemMessage
import akka.event.Logging
import akka.remote._
import akka.remote.DaemonMsgCreate
import akka.remote.QuarantinedEvent
import akka.remote.artery.AeronSink.GaveUpSendingException
import akka.remote.artery.Encoder.ChangeOutboundCompression
import akka.remote.artery.Encoder.ChangeOutboundCompressionFailed
import akka.remote.artery.InboundControlJunction.ControlMessageSubject
import akka.remote.artery.OutboundControlJunction.OutboundControlIngress
import akka.remote.artery.OutboundHandshake.HandshakeTimeoutException
import akka.remote.artery.SendQueue.ProducerApi
import akka.remote.artery.SystemMessageDelivery.ClearSystemMessageDelivery
import akka.remote.artery.compress.CompressionProtocol._
import akka.remote.artery.compress.CompressionTable
import akka.stream.AbruptTerminationException
import akka.stream.Materializer
import akka.stream.scaladsl.Keep
import akka.stream.scaladsl.Source
import akka.util.{ Unsafe, WildcardIndex }
import org.agrona.concurrent.ManyToOneConcurrentArrayQueue
import akka.util.OptionVal
import org.agrona.concurrent.ManyToOneConcurrentArrayQueue
import akka.remote.artery.compress.CompressionProtocol._
/**
@ -82,9 +84,6 @@ private[remote] class Association(
// the `SendQueue` after materialization. Using same underlying queue. This makes it possible to
// start sending (enqueuing) to the Association immediate after construction.
/** Accesses the currently active outbound compression. */
def outboundCompression: OutboundCompressions = associationState.outboundCompressions
def createQueue(capacity: Int): Queue[OutboundEnvelope] =
new ManyToOneConcurrentArrayQueue[OutboundEnvelope](capacity)
@ -93,6 +92,25 @@ private[remote] class Association(
@volatile private[this] var controlQueue: SendQueue.ProducerApi[OutboundEnvelope] = QueueWrapper(createQueue(controlQueueSize))
@volatile private[this] var _outboundControlIngress: OutboundControlIngress = _
@volatile private[this] var materializing = new CountDownLatch(1)
@volatile private[this] var changeOutboundCompression: Option[ChangeOutboundCompression] = None
def changeActorRefCompression(table: CompressionTable[ActorRef]): Future[Done] =
changeOutboundCompression match {
case Some(c) ⇒ c.changeActorRefCompression(table)
case None ⇒ Future.failed(new ChangeOutboundCompressionFailed)
}
def changeClassManifestCompression(table: CompressionTable[String]): Future[Done] =
changeOutboundCompression match {
case Some(c) ⇒ c.changeClassManifestCompression(table)
case None ⇒ Future.failed(new ChangeOutboundCompressionFailed)
}
def clearCompression(): Future[Done] =
changeOutboundCompression match {
case Some(c) ⇒ c.clearCompression()
case None ⇒ Future.failed(new ChangeOutboundCompressionFailed)
}
private val _testStages: CopyOnWriteArrayList[TestManagementApi] = new CopyOnWriteArrayList
@ -138,31 +156,41 @@ private[remote] class Association(
def associationState: AssociationState =
Unsafe.instance.getObjectVolatile(this, AbstractAssociation.sharedStateOffset).asInstanceOf[AssociationState]
def completeHandshake(peer: UniqueAddress): Unit = {
def completeHandshake(peer: UniqueAddress): Future[Done] = {
require(
remoteAddress == peer.address,
s"wrong remote address in completeHandshake, got ${peer.address}, expected $remoteAddress")
val current = associationState
current.uniqueRemoteAddressPromise.trySuccess(peer)
current.uniqueRemoteAddressValue() match {
case Some(`peer`) ⇒
// our value
if (current.outboundCompressions == NoOutboundCompressions) {
// enable outbound compression (here, since earlier we don't know the remote address)
swapState(current, current.withCompression(createOutboundCompressions(remoteAddress)))
}
// handshake already completed
Future.successful(Done)
case _ ⇒
val newState = current.newIncarnation(Promise.successful(peer), createOutboundCompressions(remoteAddress))
if (swapState(current, newState)) {
// clear outbound compression, it's safe to do that several times if someone else
// completes handshake at same time, but it's important to clear it before
// we signal that the handshake is completed (uniqueRemoteAddressPromise.trySuccess)
import transport.system.dispatcher
clearCompression().map { _ ⇒
current.uniqueRemoteAddressPromise.trySuccess(peer)
current.uniqueRemoteAddressValue() match {
case Some(old) ⇒
log.debug(
"Incarnation {} of association to [{}] with new UID [{}] (old UID [{}])",
newState.incarnation, peer.address, peer.uid, old.uid)
case None ⇒
// Failed, nothing to do
case Some(`peer`) ⇒
// our value
case _ ⇒
val newState = current.newIncarnation(Promise.successful(peer))
if (swapState(current, newState)) {
current.uniqueRemoteAddressValue() match {
case Some(old) ⇒
log.debug(
"Incarnation {} of association to [{}] with new UID [{}] (old UID [{}])",
newState.incarnation, peer.address, peer.uid, old.uid)
case None ⇒
// Failed, nothing to do
}
// if swap failed someone else completed before us, and that is fine
}
}
// if swap failed someone else completed before us, and that is fine
Done
}
}
}
@ -253,6 +281,8 @@ private[remote] class Association(
log.warning(
"Association to [{}] with UID [{}] is irrecoverably failed. Quarantining address. {}",
remoteAddress, u, reason)
// clear outbound compression
clearCompression()
// FIXME when we complete the switch to Long UID we must use Long here also, issue #20644
transport.eventPublisher.notifyListeners(QuarantinedEvent(remoteAddress, u.toInt))
// end delivery of system messages to that incarnation after this point
@ -291,20 +321,16 @@ private[remote] class Association(
}
private def runOutboundStreams(): Unit = {
// TODO no compression for control / large streams currently
val disableCompression = NoOutboundCompressions
// it's important to materialize the outboundControl stream first,
// so that outboundControlIngress is ready when stages for all streams start
runOutboundControlStream(disableCompression)
runOutboundOrdinaryMessagesStream(CurrentAssociationStateOutboundCompressionsProxy)
runOutboundControlStream()
runOutboundOrdinaryMessagesStream()
if (transport.largeMessageChannelEnabled) {
runOutboundLargeMessagesStream(disableCompression)
}
if (transport.largeMessageChannelEnabled)
runOutboundLargeMessagesStream()
}
private def runOutboundControlStream(compression: OutboundCompressions): Unit = {
private def runOutboundControlStream(): Unit = {
// stage in the control stream may access the outboundControlIngress before returned here
// using CountDownLatch to make sure that materialization is completed before accessing outboundControlIngress
materializing = new CountDownLatch(1)
@ -318,14 +344,14 @@ private[remote] class Association(
Source.fromGraph(new SendQueue[OutboundEnvelope])
.via(transport.outboundControlPart1(this))
.viaMat(transport.outboundTestFlow(this))(Keep.both)
.toMat(transport.outboundControlPart2(this, compression))(Keep.both)
.toMat(transport.outboundControlPart2(this))(Keep.both)
.run()(materializer)
_testStages.add(mgmt)
(queueValue, (control, completed))
} else {
Source.fromGraph(new SendQueue[OutboundEnvelope])
.via(transport.outboundControlPart1(this))
.toMat(transport.outboundControlPart2(this, compression))(Keep.both)
.toMat(transport.outboundControlPart2(this))(Keep.both)
.run()(materializer)
}
@ -335,7 +361,7 @@ private[remote] class Association(
_outboundControlIngress = control
materializing.countDown()
attachStreamRestart("Outbound control stream", completed, cause ⇒ {
runOutboundControlStream(compression)
runOutboundControlStream()
cause match {
case _: HandshakeTimeoutException ⇒ // ok, quarantine not possible without UID
case _ ⇒ quarantine("Outbound control stream restarted")
@ -351,32 +377,33 @@ private[remote] class Association(
QueueWrapper(createQueue(capacity))
}
private def runOutboundOrdinaryMessagesStream(compression: OutboundCompressions): Unit = {
private def runOutboundOrdinaryMessagesStream(): Unit = {
val wrapper = getOrCreateQueueWrapper(queue, queueSize)
queue = wrapper // use new underlying queue immediately for restarts
val (queueValue, completed) =
val (queueValue, (changeCompression, completed)) =
if (transport.remoteSettings.TestMode) {
val ((queueValue, mgmt), completed) = Source.fromGraph(new SendQueue[OutboundEnvelope])
.viaMat(transport.outboundTestFlow(this))(Keep.both)
.toMat(transport.outbound(this, compression))(Keep.both)
.toMat(transport.outbound(this))(Keep.both)
.run()(materializer)
_testStages.add(mgmt)
(queueValue, completed)
} else {
Source.fromGraph(new SendQueue[OutboundEnvelope])
.toMat(transport.outbound(this, compression))(Keep.both)
.toMat(transport.outbound(this))(Keep.both)
.run()(materializer)
}
queueValue.inject(wrapper.queue)
// replace with the materialized value, still same underlying queue
queue = queueValue
changeOutboundCompression = Some(changeCompression)
attachStreamRestart("Outbound message stream", completed, _ ⇒ runOutboundOrdinaryMessagesStream(compression))
attachStreamRestart("Outbound message stream", completed, _ ⇒ runOutboundOrdinaryMessagesStream())
}
private def runOutboundLargeMessagesStream(compression: OutboundCompressions): Unit = {
private def runOutboundLargeMessagesStream(): Unit = {
val wrapper = getOrCreateQueueWrapper(largeQueue, largeQueueSize)
largeQueue = wrapper // use new underlying queue immediately for restarts
@ -384,20 +411,20 @@ private[remote] class Association(
if (transport.remoteSettings.TestMode) {
val ((queueValue, mgmt), completed) = Source.fromGraph(new SendQueue[OutboundEnvelope])
.viaMat(transport.outboundTestFlow(this))(Keep.both)
.toMat(transport.outboundLarge(this, compression))(Keep.both)
.toMat(transport.outboundLarge(this))(Keep.both)
.run()(materializer)
_testStages.add(mgmt)
(queueValue, completed)
} else {
Source.fromGraph(new SendQueue[OutboundEnvelope])
.toMat(transport.outboundLarge(this, compression))(Keep.both)
.toMat(transport.outboundLarge(this))(Keep.both)
.run()(materializer)
}
queueValue.inject(wrapper.queue)
// replace with the materialized value, still same underlying queue
largeQueue = queueValue
attachStreamRestart("Outbound large message stream", completed, _ ⇒ runOutboundLargeMessagesStream(compression))
attachStreamRestart("Outbound large message stream", completed, _ ⇒ runOutboundLargeMessagesStream())
}
private def attachStreamRestart(streamName: String, streamCompleted: Future[Done], restart: Throwable ⇒ Unit): Unit = {
@ -421,41 +448,6 @@ private[remote] class Association(
}
}
// TODO: Make sure that once other channels use Compression, each gets it's own
private def createOutboundCompressions(remoteAddress: Address): OutboundCompressions = {
if (transport.provider.remoteSettings.ArteryCompressionSettings.enabled) {
val compression = new OutboundCompressionsImpl(transport.system, remoteAddress)
log.debug("Creating Outbound compression table to [{}]", remoteAddress)
compression
} else NoOutboundCompressions
}
/**
* This proxy uses the current associationStates compression table, which is reset for a new incarnation.
* This way the same outgoing stream will switch to using the new table without the need of restarting it.
*/
private object CurrentAssociationStateOutboundCompressionsProxy extends OutboundCompressions {
override def actorRefCompressionTableVersion: Int =
associationState.outboundCompressions.actorRefCompressionTableVersion
override def applyActorRefCompressionTable(table: CompressionTable[ActorRef]): Unit = {
associationState.outboundCompressions.applyActorRefCompressionTable(table)
}
override final def compressActorRef(ref: ActorRef): Int = {
associationState.outboundCompressions.compressActorRef(ref)
}
override def classManifestCompressionTableVersion: Int =
associationState.outboundCompressions.classManifestCompressionTableVersion
override def applyClassManifestCompressionTable(table: CompressionTable[String]): Unit =
associationState.outboundCompressions.applyClassManifestCompressionTable(table)
override final def compressClassManifest(manifest: String): Int =
associationState.outboundCompressions.compressClassManifest(manifest)
override def toString =
s"${Logging.simpleName(getClass)}(current delegate: ${associationState.outboundCompressions})"
}
override def toString: String =
s"Association($localAddress -> $remoteAddress with $associationState)"
@ -489,17 +481,23 @@ private[remote] class AssociationRegistry(createAssociation: Address ⇒ Associa
@tailrec final def setUID(peer: UniqueAddress): Association = {
val currentMap = associationsByUid.get
val a = association(peer.address)
// make sure we don't overwrite same UID with different association
currentMap.get(peer.uid) match {
case OptionVal.Some(previous) if (previous ne a) ⇒
throw new IllegalArgumentException(s"UID collision old [$previous] new [$a]")
case _ ⇒ // ok
case OptionVal.Some(previous) ⇒
if (previous eq a)
// associationsByUid Map already contains the right association
a
else
// make sure we don't overwrite same UID with different association
throw new IllegalArgumentException(s"UID collision old [$previous] new [$a]")
case _ ⇒
// update associationsByUid Map with the uid -> assocation
val newMap = currentMap.updated(peer.uid, a)
if (associationsByUid.compareAndSet(currentMap, newMap))
a
else
setUID(peer) // lost CAS, retry
}
val newMap = currentMap.updated(peer.uid, a)
if (associationsByUid.compareAndSet(currentMap, newMap))
a
else
setUID(peer) // lost CAS, retry
}
def allAssociations: Set[Association] =

76
akka-remote/src/main/scala/akka/remote/artery/BufferPool.scala
View file

@ -9,12 +9,13 @@ import java.nio.{ ByteBuffer, ByteOrder }
import akka.actor.{ ActorRef, Address }
import akka.remote.artery.compress.CompressionProtocol._
import akka.remote.artery.compress.{ CompressionTable, InboundCompressions, OutboundCompressions }
import akka.remote.artery.compress.{ CompressionTable, InboundCompressions }
import akka.serialization.Serialization
import org.agrona.concurrent.{ ManyToManyConcurrentArrayQueue, UnsafeBuffer }
import akka.util.{ OptionVal, Unsafe }
import scala.util.control.NonFatal
import akka.remote.artery.compress.NoInboundCompressions
/**
* INTERNAL API
@ -77,25 +78,34 @@ private[remote] object HeaderBuilder {
// We really only use the Header builder on one "side" or the other, thus in order to avoid having to split its impl
// we inject no-op compression's of the "other side".
def in(compression: InboundCompressions): HeaderBuilder = new HeaderBuilderImpl(compression, NoOutboundCompressions)
def out(compression: OutboundCompressions): HeaderBuilder = new HeaderBuilderImpl(NoInboundCompressions, compression)
def in(compression: InboundCompressions): HeaderBuilder =
new HeaderBuilderImpl(compression, CompressionTable.empty[ActorRef], CompressionTable.empty[String])
def out(): HeaderBuilder =
new HeaderBuilderImpl(NoInboundCompressions, CompressionTable.empty[ActorRef], CompressionTable.empty[String])
/** INTERNAL API, FOR TESTING ONLY */
private[remote] def bothWays(in: InboundCompressions, out: OutboundCompressions): HeaderBuilder = new HeaderBuilderImpl(in, out)
private[remote] def bothWays(
in: InboundCompressions,
outboundActorRefCompression: CompressionTable[ActorRef],
outboundClassManifestCompression: CompressionTable[String]): HeaderBuilder =
new HeaderBuilderImpl(in, outboundActorRefCompression, outboundClassManifestCompression)
}
/**
* INTERNAL API
*/
sealed trait HeaderBuilder {
private[remote] sealed trait HeaderBuilder {
def setVersion(v: Int): Unit
def version: Int
def setActorRefCompressionTableVersion(v: Int): Unit
def actorRefCompressionTableVersion: Int
def inboundActorRefCompressionTableVersion: Int
def inboundClassManifestCompressionTableVersion: Int
def setClassManifestCompressionTableVersion(v: Int): Unit
def classManifestCompressionTableVersion: Int
def outboundActorRefCompression: CompressionTable[ActorRef]
def setOutboundActorRefCompression(table: CompressionTable[ActorRef]): Unit
def outboundClassManifestCompression: CompressionTable[String]
def setOutboundClassManifestCompression(table: CompressionTable[String]): Unit
def setUid(u: Long): Unit
def uid: Long
@ -140,12 +150,15 @@ sealed trait HeaderBuilder {
/**
* INTERNAL API
*/
private[remote] final class HeaderBuilderImpl(inboundCompression: InboundCompressions, outboundCompression: OutboundCompressions) extends HeaderBuilder {
private[remote] final class HeaderBuilderImpl(
inboundCompression: InboundCompressions,
var _outboundActorRefCompression: CompressionTable[ActorRef],
var _outboundClassManifestCompression: CompressionTable[String]) extends HeaderBuilder {
// Fields only available for EnvelopeBuffer
var _version: Int = _
var _uid: Long = _
var _actorRefCompressionTableVersion: Int = 0
var _classManifestCompressionTableVersion: Int = 0
var _inboundActorRefCompressionTableVersion: Int = 0
var _inboundClassManifestCompressionTableVersion: Int = 0
var _senderActorRef: String = null
var _senderActorRefIdx: Int = -1
@ -162,14 +175,19 @@ private[remote] final class HeaderBuilderImpl(inboundCompression: InboundCompres
override def setUid(uid: Long) = _uid = uid
override def uid: Long = _uid
override def setActorRefCompressionTableVersion(v: Int): Unit = _actorRefCompressionTableVersion = v
override def actorRefCompressionTableVersion: Int = _actorRefCompressionTableVersion
override def inboundActorRefCompressionTableVersion: Int = _inboundActorRefCompressionTableVersion
override def inboundClassManifestCompressionTableVersion: Int = _inboundClassManifestCompressionTableVersion
override def setClassManifestCompressionTableVersion(v: Int): Unit = _classManifestCompressionTableVersion = v
override def classManifestCompressionTableVersion: Int = _classManifestCompressionTableVersion
def setOutboundActorRefCompression(table: CompressionTable[ActorRef]): Unit =
_outboundActorRefCompression = table
override def outboundActorRefCompression: CompressionTable[ActorRef] = _outboundActorRefCompression
def setOutboundClassManifestCompression(table: CompressionTable[String]): Unit =
_outboundClassManifestCompression = table
def outboundClassManifestCompression: CompressionTable[String] = _outboundClassManifestCompression
override def setSenderActorRef(ref: ActorRef): Unit = {
_senderActorRefIdx = outboundCompression.compressActorRef(ref)
_senderActorRefIdx = outboundActorRefCompression.compress(ref)
if (_senderActorRefIdx == -1) _senderActorRef = Serialization.serializedActorPath(ref) // includes local address from `currentTransportInformation`
}
override def setNoSender(): Unit = {
@ -179,7 +197,8 @@ private[remote] final class HeaderBuilderImpl(inboundCompression: InboundCompres
override def isNoSender: Boolean =
(_senderActorRef eq null) && _senderActorRefIdx == EnvelopeBuffer.DeadLettersCode
override def senderActorRef(originUid: Long): OptionVal[ActorRef] =
if (_senderActorRef eq null) inboundCompression.decompressActorRef(originUid, actorRefCompressionTableVersion, _senderActorRefIdx)
if (_senderActorRef eq null)
inboundCompression.decompressActorRef(originUid, inboundActorRefCompressionTableVersion, _senderActorRefIdx)
else OptionVal.None
def senderActorRefPath: OptionVal[String] =
OptionVal(_senderActorRef)
@ -192,13 +211,14 @@ private[remote] final class HeaderBuilderImpl(inboundCompression: InboundCompres
(_recipientActorRef eq null) && _recipientActorRefIdx == EnvelopeBuffer.DeadLettersCode
def setRecipientActorRef(ref: ActorRef): Unit = {
_recipientActorRefIdx = outboundCompression.compressActorRef(ref)
_recipientActorRefIdx = outboundActorRefCompression.compress(ref)
if (_recipientActorRefIdx == -1) {
_recipientActorRef = ref.path.toSerializationFormat
}
}
def recipientActorRef(originUid: Long): OptionVal[ActorRef] =
if (_recipientActorRef eq null) inboundCompression.decompressActorRef(originUid, actorRefCompressionTableVersion, _recipientActorRefIdx)
if (_recipientActorRef eq null)
inboundCompression.decompressActorRef(originUid, inboundActorRefCompressionTableVersion, _recipientActorRefIdx)
else OptionVal.None
def recipientActorRefPath: OptionVal[String] =
OptionVal(_recipientActorRef)
@ -210,13 +230,15 @@ private[remote] final class HeaderBuilderImpl(inboundCompression: InboundCompres
_serializer
override def setManifest(manifest: String): Unit = {
_manifestIdx = outboundCompression.compressClassManifest(manifest)
_manifestIdx = outboundClassManifestCompression.compress(manifest)
if (_manifestIdx == -1) _manifest = manifest
}
override def manifest(originUid: Long): String = {
if (_manifest ne null) _manifest
else {
_manifest = inboundCompression.decompressClassManifest(originUid, classManifestCompressionTableVersion, _manifestIdx).get
_manifest = inboundCompression.decompressClassManifest(
originUid,
inboundClassManifestCompressionTableVersion, _manifestIdx).get
_manifest
}
}
@ -224,8 +246,6 @@ private[remote] final class HeaderBuilderImpl(inboundCompression: InboundCompres
override def toString =
"HeaderBuilderImpl(" +
"version:" + version + ", " +
"actorRefCompressionTableVersion:" + actorRefCompressionTableVersion + ", " +
"classManifestCompressionTableVersion:" + classManifestCompressionTableVersion + ", " +
"uid:" + uid + ", " +
"_senderActorRef:" + _senderActorRef + ", " +
"_senderActorRefIdx:" + _senderActorRefIdx + ", " +
@ -257,8 +277,8 @@ private[remote] final class EnvelopeBuffer(val byteBuffer: ByteBuffer) {
byteBuffer.putInt(header.serializer)
// compression table version numbers
byteBuffer.putInt(ActorRefCompressionTableVersionTagOffset, header._actorRefCompressionTableVersion | TagTypeMask)
byteBuffer.putInt(ClassManifestCompressionTableVersionTagOffset, header._classManifestCompressionTableVersion | TagTypeMask)
byteBuffer.putInt(ActorRefCompressionTableVersionTagOffset, header.outboundActorRefCompression.version | TagTypeMask)
byteBuffer.putInt(ClassManifestCompressionTableVersionTagOffset, header.outboundClassManifestCompression.version | TagTypeMask)
// Write compressable, variable-length parts always to the actual position of the buffer
// Write tag values explicitly in their proper offset
@ -294,11 +314,11 @@ private[remote] final class EnvelopeBuffer(val byteBuffer: ByteBuffer) {
// compression table versions (stored in the Tag)
val refCompressionVersionTag = byteBuffer.getInt(ActorRefCompressionTableVersionTagOffset)
if ((refCompressionVersionTag & TagTypeMask) != 0) {
header setActorRefCompressionTableVersion refCompressionVersionTag & TagValueMask
header._inboundActorRefCompressionTableVersion = refCompressionVersionTag & TagValueMask
}
val manifestCompressionVersionTag = byteBuffer.getInt(ClassManifestCompressionTableVersionTagOffset)
if ((manifestCompressionVersionTag & TagTypeMask) != 0) {
header setClassManifestCompressionTableVersion manifestCompressionVersionTag & TagValueMask
header._inboundClassManifestCompressionTableVersion = manifestCompressionVersionTag & TagValueMask
}
// Read compressable, variable-length parts always from the actual position of the buffer

110
akka-remote/src/main/scala/akka/remote/artery/Codecs.scala
View file

@ -13,9 +13,28 @@ import akka.stream._
import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
import akka.util.{ ByteString, OptionVal, PrettyByteString }
import akka.actor.EmptyLocalActorRef
import akka.remote.artery.compress.{ InboundCompressions, OutboundCompressions, OutboundCompressionsImpl }
import akka.remote.artery.compress.InboundCompressions
import akka.stream.stage.TimerGraphStageLogic
import java.util.concurrent.TimeUnit
import scala.concurrent.Future
import akka.remote.artery.compress.CompressionTable
import akka.Done
import akka.stream.stage.GraphStageWithMaterializedValue
import scala.concurrent.Promise
/**
* INTERNAL API
*/
private[remote] object Encoder {
private[remote] trait ChangeOutboundCompression {
def changeActorRefCompression(table: CompressionTable[ActorRef]): Future[Done]
def changeClassManifestCompression(table: CompressionTable[String]): Future[Done]
def clearCompression(): Future[Done]
}
private[remote] class ChangeOutboundCompressionFailed extends RuntimeException(
"Change of outbound compression table failed (will be retried), because materialization did not complete yet")
}
/**
* INTERNAL API
@ -23,42 +42,49 @@ import java.util.concurrent.TimeUnit
private[remote] class Encoder(
uniqueLocalAddress: UniqueAddress,
system: ActorSystem,
compression: OutboundCompressions,
outboundEnvelopePool: ObjectPool[ReusableOutboundEnvelope],
bufferPool: EnvelopeBufferPool)
extends GraphStage[FlowShape[OutboundEnvelope, EnvelopeBuffer]] {
extends GraphStageWithMaterializedValue[FlowShape[OutboundEnvelope, EnvelopeBuffer], Encoder.ChangeOutboundCompression] {
import Encoder._
val in: Inlet[OutboundEnvelope] = Inlet("Artery.Encoder.in")
val out: Outlet[EnvelopeBuffer] = Outlet("Artery.Encoder.out")
val shape: FlowShape[OutboundEnvelope, EnvelopeBuffer] = FlowShape(in, out)
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) with InHandler with OutHandler with StageLogging {
override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, ChangeOutboundCompression) = {
val logic = new GraphStageLogic(shape) with InHandler with OutHandler with StageLogging with ChangeOutboundCompression {
private val headerBuilder = HeaderBuilder.out(compression)
private val headerBuilder = HeaderBuilder.out()
headerBuilder setVersion ArteryTransport.Version
headerBuilder setUid uniqueLocalAddress.uid
private val localAddress = uniqueLocalAddress.address
private val serialization = SerializationExtension(system)
private val serializationInfo = Serialization.Information(localAddress, system)
private val changeActorRefCompressionCb = getAsyncCallback[(CompressionTable[ActorRef], Promise[Done])] {
case (table, done) ⇒
headerBuilder.setOutboundActorRefCompression(table)
done.success(Done)
}
private val changeClassManifsetCompressionCb = getAsyncCallback[(CompressionTable[String], Promise[Done])] {
case (table, done) ⇒
headerBuilder.setOutboundClassManifestCompression(table)
done.success(Done)
}
private val clearCompressionCb = getAsyncCallback[Promise[Done]] { done ⇒
headerBuilder.setOutboundActorRefCompression(CompressionTable.empty[ActorRef])
headerBuilder.setOutboundClassManifestCompression(CompressionTable.empty[String])
done.success(Done)
}
override protected def logSource = classOf[Encoder]
override def onPush(): Unit = {
val outboundEnvelope = grab(in)
val envelope = bufferPool.acquire()
// FIXME: OMG race between setting the version, and using the table!!!!
// incoming messages are concurrent to outgoing ones
// incoming message may be table advertisement
// which swaps the table in Outgoing*Compression for the new one (n+1)
// by itself it does so atomically,
// race: however here we store the compression table version separately from actually using it (storing the refs / manifests etc).
// so there is a slight race IF the table is swapped right between us setting the version n here [then the table being swapped to n+1] and then we use the n+1 version to compressions the compressions (which the receiving end will fail to read, since the encoding could be completely different, and it picks the table based on the version Int).
// A solution would be to getTable => use it to set and compress things
headerBuilder setActorRefCompressionTableVersion compression.actorRefCompressionTableVersion
headerBuilder setClassManifestCompressionTableVersion compression.classManifestCompressionTableVersion
// internally compression is applied by the builder:
outboundEnvelope.recipient match {
case OptionVal.Some(r) ⇒ headerBuilder setRecipientActorRef r
@ -109,8 +135,49 @@ private[remote] class Encoder(
override def onPull(): Unit = pull(in)
/**
* External call from ChangeOutboundCompression materialized value
*/
override def changeActorRefCompression(table: CompressionTable[ActorRef]): Future[Done] = {
val done = Promise[Done]()
try changeActorRefCompressionCb.invoke((table, done)) catch {
// This is a harmless failure, it will be retried on next advertisement or handshake attempt.
// It will only occur when callback is invoked before preStart. That is highly unlikely to
// happen since advertisement is not done immediately and handshake involves network roundtrip.
case NonFatal(_) ⇒ done.tryFailure(new ChangeOutboundCompressionFailed)
}
done.future
}
/**
* External call from ChangeOutboundCompression materialized value
*/
override def changeClassManifestCompression(table: CompressionTable[String]): Future[Done] = {
val done = Promise[Done]()
try changeClassManifsetCompressionCb.invoke((table, done)) catch {
// in case materialization not completed yet
case NonFatal(_) ⇒ done.tryFailure(new ChangeOutboundCompressionFailed)
}
done.future
}
/**
* External call from ChangeOutboundCompression materialized value
*/
override def clearCompression(): Future[Done] = {
val done = Promise[Done]()
try clearCompressionCb.invoke(done) catch {
// in case materialization not completed yet
case NonFatal(_) ⇒ done.tryFailure(new ChangeOutboundCompressionFailed)
}
done.future
}
setHandlers(in, out, this)
}
(logic, logic)
}
}
/**
@ -198,20 +265,17 @@ private[remote] class Decoder(
val remoteAddress = assoc.remoteAddress
sender match {
case OptionVal.Some(snd) ⇒
compression.hitActorRef(originUid, headerBuilder.actorRefCompressionTableVersion,
remoteAddress, snd, 1)
compression.hitActorRef(originUid, remoteAddress, snd, 1)
case OptionVal.None ⇒
}
recipient match {
case OptionVal.Some(rcp) ⇒
compression.hitActorRef(originUid, headerBuilder.actorRefCompressionTableVersion,
remoteAddress, rcp, 1)
compression.hitActorRef(originUid, remoteAddress, rcp, 1)
case OptionVal.None ⇒
}
compression.hitClassManifest(originUid, headerBuilder.classManifestCompressionTableVersion,
remoteAddress, classManifest, 1)
compression.hitClassManifest(originUid, remoteAddress, classManifest, 1)
case _ ⇒
// we don't want to record hits for compression while handshake is still in progress.

44
akka-remote/src/main/scala/akka/remote/artery/Handshake.scala
View file

@ -18,6 +18,8 @@ import akka.stream.stage.InHandler
import akka.stream.stage.OutHandler
import akka.stream.stage.TimerGraphStageLogic
import akka.util.OptionVal
import akka.Done
import scala.concurrent.Future
/**
* INTERNAL API
@ -177,8 +179,9 @@ private[akka] class InboundHandshake(inboundContext: InboundContext, inControlSt
env.message match {
case HandshakeReq(from) ⇒ onHandshakeReq(from)
case HandshakeRsp(from) ⇒
inboundContext.completeHandshake(from)
pull(in)
after(inboundContext.completeHandshake(from)) {
pull(in)
}
case _ ⇒
onMessage(env)
}
@ -197,21 +200,32 @@ private[akka] class InboundHandshake(inboundContext: InboundContext, inControlSt
})
private def onHandshakeReq(from: UniqueAddress): Unit = {
inboundContext.completeHandshake(from)
inboundContext.sendControl(from.address, HandshakeRsp(inboundContext.localAddress))
pull(in)
after(inboundContext.completeHandshake(from)) {
inboundContext.sendControl(from.address, HandshakeRsp(inboundContext.localAddress))
pull(in)
}
}
private def after(first: Future[Done])(thenInside: ⇒ Unit): Unit = {
first.value match {
case Some(_) ⇒
// This in the normal case (all but the first). The future will be completed
// because handshake was already completed. Note that we send those HandshakeReq
// periodically.
thenInside
case None ⇒
implicit val ec = materializer.executionContext
first.onComplete { _ ⇒
getAsyncCallback[Done](_ ⇒ thenInside).invoke(Done)
}
}
}
private def onMessage(env: InboundEnvelope): Unit = {
if (isKnownOrigin(env))
push(out, env)
else {
// FIXME remove, only debug
log.warning(
s"Dropping message [{}] from unknown system with UID [{}]. " +
"This system with UID [{}] was probably restarted. " +
"Messages will be accepted when new handshake has been completed.",
env.message.getClass.getName, inboundContext.localAddress.uid, env.originUid)
if (log.isDebugEnabled)
log.debug(
s"Dropping message [{}] from unknown system with UID [{}]. " +
@ -222,8 +236,12 @@ private[akka] class InboundHandshake(inboundContext: InboundContext, inControlSt
}
}
private def isKnownOrigin(env: InboundEnvelope): Boolean =
env.association.isDefined
private def isKnownOrigin(env: InboundEnvelope): Boolean = {
// the association is passed in the envelope from the Decoder stage to avoid
// additional lookup. The second OR case is because if we didn't use fusing it
// would be possible that it was not found by Decoder (handshake not completed yet)
env.association.isDefined || inboundContext.association(env.originUid).isDefined
}
// OutHandler
override def onPull(): Unit = pull(in)

98
akka-remote/src/main/scala/akka/remote/artery/compress/ActualCompressions.scala
View file

@ -1,98 +0,0 @@
/**
* Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
*/
package akka.remote.artery.compress
import java.util.function.LongFunction
import akka.actor.{ ActorRef, ActorSystem, Address }
import akka.remote.artery._
import akka.util.OptionVal
import org.agrona.collections.Long2ObjectHashMap
/** INTERNAL API */
private[remote] final class OutboundCompressionsImpl(system: ActorSystem, remoteAddress: Address) extends OutboundCompressions {
private val actorRefsOut = new OutboundActorRefCompression(system, remoteAddress)
private val classManifestsOut = new OutboundClassManifestCompression(system, remoteAddress)
// actor ref compression ---
override def compressActorRef(ref: ActorRef): Int = actorRefsOut.compress(ref)
override def actorRefCompressionTableVersion: Int = actorRefsOut.activeCompressionTableVersion
override def applyActorRefCompressionTable(table: CompressionTable[ActorRef]): Unit =
actorRefsOut.flipTable(table)
// class manifest compression ---
override def compressClassManifest(manifest: String): Int = classManifestsOut.compress(manifest)
override def classManifestCompressionTableVersion: Int = classManifestsOut.activeCompressionTableVersion
override def applyClassManifestCompressionTable(table: CompressionTable[String]): Unit =
classManifestsOut.flipTable(table)
}
/**
* INTERNAL API
*
* One per incoming Aeron stream, actual compression tables are kept per-originUid and created on demand.
*/
private[remote] final class InboundCompressionsImpl(
system: ActorSystem,
inboundContext: InboundContext) extends InboundCompressions {
private val settings = CompressionSettings(system)
// FIXME we also must remove the ones that won't be used anymore - when quarantine triggers
private[this] val _actorRefsIns = new Long2ObjectHashMap[InboundActorRefCompression]()
private val createInboundActorRefsForOrigin = new LongFunction[InboundActorRefCompression] {
override def apply(originUid: Long): InboundActorRefCompression = {
val actorRefHitters = new TopHeavyHitters[ActorRef](settings.actorRefs.max)
new InboundActorRefCompression(system, settings, originUid, inboundContext, actorRefHitters)
}
}
private def actorRefsIn(originUid: Long): InboundActorRefCompression =
_actorRefsIns.computeIfAbsent(originUid, createInboundActorRefsForOrigin)
private[this] val _classManifestsIns = new Long2ObjectHashMap[InboundManifestCompression]()
private val createInboundManifestsForOrigin = new LongFunction[InboundManifestCompression] {
override def apply(originUid: Long): InboundManifestCompression = {
val manifestHitters = new TopHeavyHitters[String](settings.manifests.max)
new InboundManifestCompression(system, settings, originUid, inboundContext, manifestHitters)
}
}
private def classManifestsIn(originUid: Long): InboundManifestCompression =
_classManifestsIns.computeIfAbsent(originUid, createInboundManifestsForOrigin)
// actor ref compression ---
override def decompressActorRef(originUid: Long, tableVersion: Int, idx: Int): OptionVal[ActorRef] =
actorRefsIn(originUid).decompress(tableVersion, idx)
override def hitActorRef(originUid: Long, tableVersion: Int, address: Address, ref: ActorRef, n: Int): Unit =
actorRefsIn(originUid).increment(address, ref, n)
override def confirmActorRefCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit =
actorRefsIn(originUid).confirmAdvertisement(tableVersion)
// class manifest compression ---
override def decompressClassManifest(originUid: Long, tableVersion: Int, idx: Int): OptionVal[String] =
classManifestsIn(originUid).decompress(tableVersion, idx)
override def hitClassManifest(originUid: Long, tableVersion: Int, address: Address, manifest: String, n: Int): Unit =
classManifestsIn(originUid).increment(address, manifest, n)
override def confirmClassManifestCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit =
actorRefsIn(originUid).confirmAdvertisement(tableVersion)
// testing utilities ---
/** INTERNAL API: for testing only */
private[remote] def runNextActorRefAdvertisement() = {
import scala.collection.JavaConverters._
_actorRefsIns.values().asScala.foreach { inbound ⇒ inbound.runNextTableAdvertisement() }
}
/** INTERNAL API: for testing only */
private[remote] def runNextClassManifestAdvertisement() = {
import scala.collection.JavaConverters._
_classManifestsIns.values().asScala.foreach { inbound ⇒ inbound.runNextTableAdvertisement() }
}
}

39
akka-remote/src/main/scala/akka/remote/artery/compress/AllCompressions.scala
View file

@ -1,39 +0,0 @@
/*
* Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
*/
package akka.remote.artery.compress
import akka.actor.{ ActorRef, Address }
import akka.util.OptionVal
/**
* INTERNAL API
* Decompress and cause compression advertisements.
*
* One per inbound message stream thus must demux by originUid to use the right tables.
*/
private[remote] trait InboundCompressions {
def hitActorRef(originUid: Long, tableVersion: Int, remote: Address, ref: ActorRef, n: Int): Unit
def decompressActorRef(originUid: Long, tableVersion: Int, idx: Int): OptionVal[ActorRef]
def confirmActorRefCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit
def hitClassManifest(originUid: Long, tableVersion: Int, remote: Address, manifest: String, n: Int): Unit
def decompressClassManifest(originUid: Long, tableVersion: Int, idx: Int): OptionVal[String]
def confirmClassManifestCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit
}
/**
* INTERNAL API
* Compress outgoing data and handle compression advertisements to fill compression table.
*
* One per outgoing message stream.
*/
private[remote] trait OutboundCompressions {
def applyActorRefCompressionTable(table: CompressionTable[ActorRef]): Unit
def actorRefCompressionTableVersion: Int
def compressActorRef(ref: ActorRef): Int
def applyClassManifestCompressionTable(table: CompressionTable[String]): Unit
def classManifestCompressionTableVersion: Int
def compressClassManifest(manifest: String): Int
}

9
akka-remote/src/main/scala/akka/remote/artery/compress/CompressionTable.scala
View file

@ -6,6 +6,13 @@ package akka.remote.artery.compress
/** INTERNAL API: Versioned compression table to be advertised between systems */
private[akka] final case class CompressionTable[T](version: Int, map: Map[T, Int]) {
import CompressionTable.NotCompressedId
def compress(value: T): Int =
map.get(value) match {
case Some(id) ⇒ id
case None ⇒ NotCompressedId
}
def invert: DecompressionTable[T] =
if (map.isEmpty) DecompressionTable.empty[T].copy(version = version)
@ -25,6 +32,8 @@ private[akka] final case class CompressionTable[T](version: Int, map: Map[T, Int
}
/** INTERNAL API */
private[remote] object CompressionTable {
final val NotCompressedId = -1
private[this] val _empty = new CompressionTable[Any](0, Map.empty)
def empty[T] = _empty.asInstanceOf[CompressionTable[T]]
}

91
akka-remote/src/main/scala/akka/remote/artery/compress/InboundCompressions.scala
View file

@ -4,12 +4,97 @@
package akka.remote.artery.compress
import java.util.concurrent.atomic.AtomicReference
import java.util.function.LongFunction
import scala.concurrent.duration.{ Duration, FiniteDuration }
import akka.actor.{ ActorRef, ActorSystem, Address }
import akka.event.{ Logging, NoLogging }
import akka.remote.artery.{ InboundContext, OutboundContext }
import akka.util.{ OptionVal, PrettyDuration }
import scala.concurrent.duration.{ Duration, FiniteDuration }
import java.util.concurrent.atomic.AtomicReference
import org.agrona.collections.Long2ObjectHashMap
/**
* INTERNAL API
* Decompress and cause compression advertisements.
*
* One per inbound message stream thus must demux by originUid to use the right tables.
*/
private[remote] trait InboundCompressions {
def hitActorRef(originUid: Long, remote: Address, ref: ActorRef, n: Int): Unit
def decompressActorRef(originUid: Long, tableVersion: Int, idx: Int): OptionVal[ActorRef]
def confirmActorRefCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit
def hitClassManifest(originUid: Long, remote: Address, manifest: String, n: Int): Unit
def decompressClassManifest(originUid: Long, tableVersion: Int, idx: Int): OptionVal[String]
def confirmClassManifestCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit
}
/**
* INTERNAL API
*
* One per incoming Aeron stream, actual compression tables are kept per-originUid and created on demand.
*/
private[remote] final class InboundCompressionsImpl(
system: ActorSystem,
inboundContext: InboundContext) extends InboundCompressions {
private val settings = CompressionSettings(system)
// FIXME we also must remove the ones that won't be used anymore - when quarantine triggers
private[this] val _actorRefsIns = new Long2ObjectHashMap[InboundActorRefCompression]()
private val createInboundActorRefsForOrigin = new LongFunction[InboundActorRefCompression] {
override def apply(originUid: Long): InboundActorRefCompression = {
val actorRefHitters = new TopHeavyHitters[ActorRef](settings.actorRefs.max)
new InboundActorRefCompression(system, settings, originUid, inboundContext, actorRefHitters)
}
}
private def actorRefsIn(originUid: Long): InboundActorRefCompression =
_actorRefsIns.computeIfAbsent(originUid, createInboundActorRefsForOrigin)
private[this] val _classManifestsIns = new Long2ObjectHashMap[InboundManifestCompression]()
private val createInboundManifestsForOrigin = new LongFunction[InboundManifestCompression] {
override def apply(originUid: Long): InboundManifestCompression = {
val manifestHitters = new TopHeavyHitters[String](settings.manifests.max)
new InboundManifestCompression(system, settings, originUid, inboundContext, manifestHitters)
}
}
private def classManifestsIn(originUid: Long): InboundManifestCompression =
_classManifestsIns.computeIfAbsent(originUid, createInboundManifestsForOrigin)
// actor ref compression ---
override def decompressActorRef(originUid: Long, tableVersion: Int, idx: Int): OptionVal[ActorRef] =
actorRefsIn(originUid).decompress(tableVersion, idx)
override def hitActorRef(originUid: Long, address: Address, ref: ActorRef, n: Int): Unit =
actorRefsIn(originUid).increment(address, ref, n)
override def confirmActorRefCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit =
actorRefsIn(originUid).confirmAdvertisement(tableVersion)
// class manifest compression ---
override def decompressClassManifest(originUid: Long, tableVersion: Int, idx: Int): OptionVal[String] =
classManifestsIn(originUid).decompress(tableVersion, idx)
override def hitClassManifest(originUid: Long, address: Address, manifest: String, n: Int): Unit =
classManifestsIn(originUid).increment(address, manifest, n)
override def confirmClassManifestCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit =
actorRefsIn(originUid).confirmAdvertisement(tableVersion)
// testing utilities ---
/** INTERNAL API: for testing only */
private[remote] def runNextActorRefAdvertisement() = {
import scala.collection.JavaConverters._
_actorRefsIns.values().asScala.foreach { inbound ⇒ inbound.runNextTableAdvertisement() }
}
/** INTERNAL API: for testing only */
private[remote] def runNextClassManifestAdvertisement() = {
import scala.collection.JavaConverters._
_classManifestsIns.values().asScala.foreach { inbound ⇒ inbound.runNextTableAdvertisement() }
}
}
/**
* INTERNAL API
@ -105,7 +190,7 @@ private[remote] abstract class InboundCompression[T >: Null](
lazy val log = Logging(system, getClass.getSimpleName)
// TODO NOTE: there exist edge cases around, we advertise table 1, accumulate table 2, the remote system has not used 2 yet,
// FIXME NOTE: there exist edge cases around, we advertise table 1, accumulate table 2, the remote system has not used 2 yet,
// yet we technically could already prepare table 3, then it starts using table 1 suddenly. Edge cases like that.
// SOLUTION 1: We don't start building new tables until we've seen the previous one be used (move from new to active)
// This is nice as it practically disables all the "build the table" work when the other side is not interested in using it.

21
akka-remote/src/main/scala/akka/remote/artery/NoLiteralCompression.scala → akka-remote/src/main/scala/akka/remote/artery/compress/NoInboundCompressions.scala
View file

@ -1,10 +1,9 @@
/**
* Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
*/
package akka.remote.artery
package akka.remote.artery.compress
import akka.actor.{ ActorRef, Address }
import akka.remote.artery.compress.{ CompressionTable, InboundCompressions, OutboundCompressions }
import akka.util.OptionVal
/**
@ -13,30 +12,16 @@ import akka.util.OptionVal
* Literarily, no compression!
*/
case object NoInboundCompressions extends InboundCompressions {
override def hitActorRef(originUid: Long, tableVersion: Int, remote: Address, ref: ActorRef, n: Int): Unit = ()
override def hitActorRef(originUid: Long, remote: Address, ref: ActorRef, n: Int): Unit = ()
override def decompressActorRef(originUid: Long, tableVersion: Int, idx: Int): OptionVal[ActorRef] =
if (idx == -1) throw new IllegalArgumentException("Attemted decompression of illegal compression id: -1")
else OptionVal.None
override def confirmActorRefCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit = ()
override def hitClassManifest(originUid: Long, tableVersion: Int, remote: Address, manifest: String, n: Int): Unit = ()
override def hitClassManifest(originUid: Long, remote: Address, manifest: String, n: Int): Unit = ()
override def decompressClassManifest(originUid: Long, tableVersion: Int, idx: Int): OptionVal[String] =
if (idx == -1) throw new IllegalArgumentException("Attemted decompression of illegal compression id: -1")
else OptionVal.None
override def confirmClassManifestCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit = ()
}
/**
* INTERNAL API
*
* Literarily, no compression!
*/
case object NoOutboundCompressions extends OutboundCompressions {
override def applyActorRefCompressionTable(table: CompressionTable[ActorRef]): Unit = ()
override def actorRefCompressionTableVersion: Int = 0
override def compressActorRef(ref: ActorRef): Int = -1
override def applyClassManifestCompressionTable(table: CompressionTable[String]): Unit = ()
override def classManifestCompressionTableVersion: Int = 0
override def compressClassManifest(manifest: String): Int = -1
}

114
akka-remote/src/main/scala/akka/remote/artery/compress/OutboundActorRefCompression.scala
View file

@ -1,114 +0,0 @@
/*
* Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
*/
package akka.remote.artery.compress
import java.util.concurrent.atomic.AtomicReference
import java.{ util ⇒ ju }
import akka.actor.{ ActorRef, ActorSystem, Address }
import akka.event.{ Logging, LoggingAdapter }
import akka.remote.artery.compress.OutboundCompression.OutboundCompressionState
import scala.annotation.tailrec
/** INTERNAL API */
private[remote] final class OutboundActorRefCompression(system: ActorSystem, remoteAddress: Address)
extends OutboundCompressionTable[ActorRef](system, remoteAddress) {
flipTable(CompressionTable(
version = 0,
map = Map(
system.deadLetters → 0)))
}
/** INTERNAL API */
private[remote] final class OutboundClassManifestCompression(system: ActorSystem, remoteAddress: Address)
extends OutboundCompressionTable[String](system, remoteAddress) {
flipTable(CompressionTable(version = 0, Map.empty))
}
/**
* INTERNAL API
* Base class for all outgoing compression.
* Encapsulates the compressedId registration and lookup.
*/
private[remote] class OutboundCompressionTable[T](system: ActorSystem, remoteAddress: Address)
extends AtomicReference[OutboundCompressionState[T]](OutboundCompressionState.initial) { // TODO could be instead via Unsafe
import OutboundCompression._
// TODO: The compression map may benefit from padding if we want multiple compressions to be running in parallel
protected val log: LoggingAdapter = Logging(system, Logging.simpleName(getClass))
// TODO this exposes us to a race between setting the Version and USING the table...?
def activeCompressionTableVersion = {
val version = get.version
version
}
/**
* Flips the currently used compression table to the new one (iff the new one has a version number higher than the currently used one).
*/
// (╯°□°）╯︵ ┻━┻
@tailrec final def flipTable(activate: CompressionTable[T]): Unit = {
val state = get()
if (activate.version > state.version) // TODO this should handle roll-over as we move to Byte
if (compareAndSet(state, prepareState(activate)))
log.debug(s"Successfully flipped compression table versions {}=>{}, for outgoing to [{}]", state.version, activate.version, remoteAddress)
else
flipTable(activate) // retry
else if (state.version == activate.version)
log.debug("Received duplicate compression table (version: {})! Ignoring it.", state.version)
else
log.error("Received unexpected compression table with version nr [{}]! " +
"Current version number is [{}].", activate.version, state.version)
}
// TODO this is crazy hot-path; optimised FastUtil-like Object->int hash map would perform better here (and avoid Integer) allocs
final def compress(value: T): Int =
get().table.getOrDefault(value, NotCompressedId)
private final def prepareState(activate: CompressionTable[T]): OutboundCompressionState[T] = {
val size = activate.map.size
// load factor is `1` since we will never grow this table beyond the initial size,
// this way we can avoid any rehashing from happening.
val m = new ju.HashMap[T, Integer](size, 1.0f) // TODO could be replaced with primitive `int` specialized version
activate.map.foreach {
case (key, value) ⇒ m.put(key, value) // TODO boxing :<
}
OutboundCompressionState(activate.version, m)
}
def toDebugString: String = {
s"""${Logging.simpleName(getClass)}(
| version: ${get.version} to [$remoteAddress]
| ${get.table}
|)""".stripMargin
}
override def toString = {
val s = get
s"""${Logging.simpleName(getClass)}(to: $remoteAddress, version: ${s.version}, compressedEntries: ${s.table.size})"""
}
}
/** INTERNAL API */
private[remote] object OutboundCompression {
// format: OFF
final val DeadLettersId = 0
final val NotCompressedId = -1
// format: ON
/** INTERNAL API */
private[remote] final case class OutboundCompressionState[T](version: Int, table: ju.Map[T, Integer])
private[remote] object OutboundCompressionState {
def initial[T] = OutboundCompressionState[T](-1, ju.Collections.emptyMap())
}
}

30
akka-remote/src/test/scala/akka/remote/artery/EnvelopeBufferSpec.scala
View file

@ -7,14 +7,14 @@ package akka.remote.artery
import java.nio.{ ByteBuffer, ByteOrder }
import akka.actor._
import akka.remote.artery.compress.{ CompressionTable, CompressionTestUtils, InboundCompressions, OutboundCompressions }
import akka.remote.artery.compress.{ CompressionTable, CompressionTestUtils, InboundCompressions }
import akka.testkit.AkkaSpec
import akka.util.{ ByteString, OptionVal }
class EnvelopeBufferSpec extends AkkaSpec {
import CompressionTestUtils._
object TestCompressor extends InboundCompressions with OutboundCompressions {
object TestCompressor extends InboundCompressions {
val refToIdx: Map[ActorRef, Int] = Map(
minimalRef("compressable0") → 0,
minimalRef("compressable1") → 1,
@ -31,24 +31,24 @@ class EnvelopeBufferSpec extends AkkaSpec {
"manifest1" → 1)
val idxToManifest = manifestToIdx.map(_.swap)
override def applyActorRefCompressionTable(table: CompressionTable[ActorRef]): Unit = ??? // dynamic allocating not needed in these tests
override def actorRefCompressionTableVersion: Int = 0
override def compressActorRef(ref: ActorRef): Int = refToIdx.getOrElse(ref, -1)
override def hitActorRef(originUid: Long, tableVersion: Int, remote: Address, ref: ActorRef, n: Int): Unit = ()
val outboundActorRefTable: CompressionTable[ActorRef] =
CompressionTable(version = 0xCAFE, refToIdx)
val outboundClassManifestTable: CompressionTable[String] =
CompressionTable(version = 0xBABE, manifestToIdx)
override def hitActorRef(originUid: Long, remote: Address, ref: ActorRef, n: Int): Unit = ()
override def decompressActorRef(originUid: Long, tableVersion: Int, idx: Int): OptionVal[ActorRef] = OptionVal(idxToRef(idx))
override def confirmActorRefCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit = ()
override def applyClassManifestCompressionTable(table: CompressionTable[String]): Unit = ??? // dynamic allocating not needed in these tests
override def classManifestCompressionTableVersion: Int = 0
override def compressClassManifest(manifest: String): Int = manifestToIdx.getOrElse(manifest, -1)
override def hitClassManifest(originUid: Long, tableVersion: Int, remote: Address, manifest: String, n: Int): Unit = ()
override def hitClassManifest(originUid: Long, remote: Address, manifest: String, n: Int): Unit = ()
override def decompressClassManifest(originUid: Long, tableVersion: Int, idx: Int): OptionVal[String] = OptionVal(idxToManifest(idx))
override def confirmClassManifestCompressionAdvertisement(originUid: Long, tableVersion: Int): Unit = ()
}
"EnvelopeBuffer" must {
val headerIn = HeaderBuilder.bothWays(TestCompressor, TestCompressor)
val headerOut = HeaderBuilder.bothWays(TestCompressor, TestCompressor)
val headerIn = HeaderBuilder.bothWays(TestCompressor, TestCompressor.outboundActorRefTable, TestCompressor.outboundClassManifestTable)
val headerOut = HeaderBuilder.bothWays(TestCompressor, TestCompressor.outboundActorRefTable, TestCompressor.outboundClassManifestTable)
val byteBuffer = ByteBuffer.allocate(1024).order(ByteOrder.LITTLE_ENDIAN)
val envelope = new EnvelopeBuffer(byteBuffer)
@ -59,8 +59,6 @@ class EnvelopeBufferSpec extends AkkaSpec {
headerIn setVersion 1
headerIn setUid 42
headerIn setSerializer 4
headerIn setActorRefCompressionTableVersion 0xCAFE
headerIn setClassManifestCompressionTableVersion 0xBABE
headerIn setRecipientActorRef minimalRef("compressable1")
headerIn setSenderActorRef minimalRef("compressable0")
@ -74,8 +72,8 @@ class EnvelopeBufferSpec extends AkkaSpec {
headerOut.version should ===(1)
headerOut.uid should ===(42)
headerOut.actorRefCompressionTableVersion should ===(0xCAFE)
headerOut.classManifestCompressionTableVersion should ===(0xBABE)
headerOut.inboundActorRefCompressionTableVersion should ===(0xCAFE)
headerOut.inboundClassManifestCompressionTableVersion should ===(0xBABE)
headerOut.serializer should ===(4)
headerOut.senderActorRef(originUid).get.path.toSerializationFormat should ===("akka://EnvelopeBufferSpec/compressable0")
headerOut.senderActorRefPath should ===(OptionVal.None)

15
akka-remote/src/test/scala/akka/remote/artery/TestContext.scala
View file

@ -18,6 +18,7 @@ import akka.remote.artery.InboundControlJunction.ControlMessageObserver
import akka.remote.artery.InboundControlJunction.ControlMessageSubject
import akka.util.OptionVal
import akka.actor.InternalActorRef
import akka.dispatch.ExecutionContexts
private[akka] class TestInboundContext(
override val localAddress: UniqueAddress,
@ -47,10 +48,13 @@ private[akka] class TestInboundContext(
override def association(uid: Long): OptionVal[OutboundContext] =
OptionVal(associationsByUid.get(uid))
override def completeHandshake(peer: UniqueAddress): Unit = {
override def completeHandshake(peer: UniqueAddress): Future[Done] = {
val a = association(peer.address).asInstanceOf[TestOutboundContext]
a.completeHandshake(peer)
associationsByUid.put(peer.uid, a)
val done = a.completeHandshake(peer)
done.foreach { _ ⇒
associationsByUid.put(peer.uid, a)
}(ExecutionContexts.sameThreadExecutionContext)
done
}
protected def createAssociation(remoteAddress: Address): TestOutboundContext =
@ -70,13 +74,14 @@ private[akka] class TestOutboundContext(
_associationState
}
def completeHandshake(peer: UniqueAddress): Unit = synchronized {
def completeHandshake(peer: UniqueAddress): Future[Done] = synchronized {
_associationState.uniqueRemoteAddressPromise.trySuccess(peer)
_associationState.uniqueRemoteAddress.value match {
case Some(Success(`peer`)) ⇒ // our value
case _ ⇒
_associationState = _associationState.newIncarnation(Promise.successful(peer), NoOutboundCompressions)
_associationState = _associationState.newIncarnation(Promise.successful(peer))
}
Future.successful(Done)
}
override def quarantine(reason: String): Unit = synchronized {

29
akka-remote/src/test/scala/akka/remote/artery/compress/OutboundCompressionSpec.scala
View file

@ -10,40 +10,23 @@ import akka.testkit.AkkaSpec
class OutboundCompressionSpec extends AkkaSpec {
import CompressionTestUtils._
val remoteAddress = Address("artery", "example", "localhost", 0)
"OutboundCompression" must {
"not compress not-known values" in {
val table = new OutboundActorRefCompression(system, remoteAddress)
table.compress(minimalRef("banana")) should ===(-1)
}
}
"OutboundActorRefCompression" must {
"Outbound ActorRef compression" must {
val alice = minimalRef("alice")
val bob = minimalRef("bob")
"always compress /deadLetters" in {
val table = new OutboundActorRefCompression(system, remoteAddress)
table.compress(system.deadLetters) should ===(0)
}
"not compress unknown actor ref" in {
val table = new OutboundActorRefCompression(system, remoteAddress)
val table = CompressionTable.empty[ActorRef]
table.compress(alice) should ===(-1) // not compressed
}
"compress previously registered actor ref" in {
val compression = new OutboundActorRefCompression(system, remoteAddress)
val table = CompressionTable(1, Map(system.deadLetters → 0, alice → 1))
compression.flipTable(table)
compression.compress(alice) should ===(1) // compressed
compression.compress(bob) should ===(-1) // not compressed
table.compress(alice) should ===(1) // compressed
table.compress(bob) should ===(-1) // not compressed
val table2 = table.copy(2, map = table.map.updated(bob, 2))
compression.flipTable(table2)
compression.compress(alice) should ===(1) // compressed
compression.compress(bob) should ===(2) // compressed
table2.compress(alice) should ===(1) // compressed
table2.compress(bob) should ===(2) // compressed
}
}