Remoting (codename Artery)

Canonical address

In order to remoting to work properly, where each system can send messages to any other system on the same network (for example a system forwards a message to a third system, and the third replies directly to the sender system) it is essential for every system to have a unique, globally reachable address and port. This address is part of the unique name of the system and will be used by other systems to open a connection to it and send messages. This means that if a host has multiple names (different DNS records pointing to the same IP address) then only one of these can be canonical. If a message arrives to a system but it contains a different hostname than the expected canonical name then the message will be dropped. If multiple names for a system would be allowed, then equality checks among ActorRef instances would no longer to be trusted and this would violate the fundamental assumption that an actor has a globally unique reference on a given network. As a consequence, this also means that localhost addresses (e.g. 127.0.0.1) cannot be used in general (apart from local development) since they are not unique addresses in a real network.

In cases, where Network Address Translation (NAT) is used or other network bridging is involved, it is important to configure the system so that it understands that there is a difference between his externally visible, canonical address and between the host-port pair that is used to listen for connections. See Akka behind NAT or in a Docker container for details.

Looking up Remote Actors

actorSelection(path) will obtain an ActorSelection to an Actor on a remote node, e.g.:

val selection =
  context.actorSelection("akka://actorSystemName@10.0.0.1:25520/user/actorName")

As you can see from the example above the following pattern is used to find an actor on a remote node:

akka://<actor system>@<hostname>:<port>/<actor path>

Once you obtained a selection to the actor you can interact with it in the same way you would with a local actor, e.g.:

selection ! "Pretty awesome feature"

To acquire an ActorRef for an ActorSelection you need to send a message to the selection and use the sender reference of the reply from the actor. There is a built-in Identify message that all Actors will understand and automatically reply to with a ActorIdentity message containing the ActorRef. This can also be done with the resolveOne method of the ActorSelection, which returns a Future of the matching ActorRef.

For more details on how actor addresses and paths are formed and used, please refer to アクターの参照、パス、アドレス.

Creating Actors Remotely

If you want to use the creation functionality in Akka remoting you have to further amend the application.conf file in the following way (only showing deployment section):

akka {
  actor {
    deployment {
      /sampleActor {
        remote = "akka://sampleActorSystem@127.0.0.1:2553"
      }
    }
  }
}

The configuration above instructs Akka to react when an actor with path /sampleActor is created, i.e. using system.actorOf(Props(...), "sampleActor"). This specific actor will not be directly instantiated, but instead the remote daemon of the remote system will be asked to create the actor, which in this sample corresponds to sampleActorSystem@127.0.0.1:2553.

Once you have configured the properties above you would do the following in code:

ActorRef actor = system.actorOf(Props.create(SampleActor.class), "sampleActor");
actor.tell("Pretty slick", ActorRef.noSender());

The actor class SampleActor has to be available to the runtimes using it, i.e. the classloader of the actor systems has to have a JAR containing the class.

You can use asterisks as wildcard matches for the actor paths, so you could specify: /*/sampleActor and that would match all sampleActor on that level in the hierarchy. You can also use wildcard in the last position to match all actors at a certain level: /someParent/*. Non-wildcard matches always have higher priority to match than wildcards, so: /foo/bar is considered more specific than /foo/* and only the highest priority match is used. Please note that it cannot be used to partially match section, like this: /foo*/bar, /f*o/bar etc.

Programmatic Remote Deployment

To allow dynamically deployed systems, it is also possible to include deployment configuration in the Props which are used to create an actor: this information is the equivalent of a deployment section from the configuration file, and if both are given, the external configuration takes precedence.

With these imports:

import akka.actor.ActorRef;
import akka.actor.Address;
import akka.actor.AddressFromURIString;
import akka.actor.Deploy;
import akka.actor.Props;
import akka.actor.ActorSystem;
import akka.remote.RemoteScope;

and a remote address like this:

Address addr = new Address("akka", "sys", "host", 1234);
addr = AddressFromURIString.parse("akka://sys@host:1234"); // the same

you can advise the system to create a child on that remote node like so:

ActorRef ref = system.actorOf(Props.create(SampleActor.class).withDeploy(
  new Deploy(new RemoteScope(addr))));

Untrusted Mode

As soon as an actor system can connect to another remotely, it may in principle send any possible message to any actor contained within that remote system. One example may be sending a PoisonPill to the system guardian, shutting that system down. This is not always desired, and it can be disabled with the following setting:

akka.remote.artery.untrusted-mode = on

This disallows sending of system messages (actor life-cycle commands, DeathWatch, etc.) and any message extending PossiblyHarmful to the system on which this flag is set. Should a client send them nonetheless they are dropped and logged (at DEBUG level in order to reduce the possibilities for a denial of service attack). PossiblyHarmful covers the predefined messages like PoisonPill and Kill, but it can also be added as a marker trait to user-defined messages.

Messages sent with actor selection are by default discarded in untrusted mode, but permission to receive actor selection messages can be granted to specific actors defined in configuration:

akka.remote.artery..trusted-selection-paths = ["/user/receptionist", "/user/namingService"]

The actual message must still not be of type PossiblyHarmful.

In summary, the following operations are ignored by a system configured in untrusted mode when incoming via the remoting layer:

• remote deployment (which also means no remote supervision)
• remote DeathWatch
• system.stop(), PoisonPill, Kill
• sending any message which extends from the PossiblyHarmful marker interface, which includes Terminated
• messages sent with actor selection, unless destination defined in trusted-selection-paths.

Watching Remote Actors

Watching a remote actor is API wise not different than watching a local actor, as described in ライフサイクルの監視、DeathWatch. However, it is important to note, that unlike in the local case, remoting has to handle when a remote actor does not terminate in a graceful way sending a system message to notify the watcher actor about the event, but instead being hosted on a system which stopped abruptly (crashed). These situations are handled by the built-in failure detector.

Failure Detector

Under the hood remote death watch uses heartbeat messages and a failure detector to generate Terminated message from network failures and JVM crashes, in addition to graceful termination of watched actor.

The heartbeat arrival times is interpreted by an implementation of The Phi Accrual Failure Detector.

The suspicion level of failure is given by a value called phi. The basic idea of the phi failure detector is to express the value of phi on a scale that is dynamically adjusted to reflect current network conditions.

The value of phi is calculated as:

phi = -log10(1 - F(timeSinceLastHeartbeat))

where F is the cumulative distribution function of a normal distribution with mean and standard deviation estimated from historical heartbeat inter-arrival times.

In the Remote Configuration you can adjust the akka.remote.watch-failure-detector.threshold to define when a phi value is considered to be a failure.

A low threshold is prone to generate many false positives but ensures a quick detection in the event of a real crash. Conversely, a high threshold generates fewer mistakes but needs more time to detect actual crashes. The default threshold is 10 and is appropriate for most situations. However in cloud environments, such as Amazon EC2, the value could be increased to 12 in order to account for network issues that sometimes occur on such platforms.

The following chart illustrates how phi increase with increasing time since the previous heartbeat.

Phi is calculated from the mean and standard deviation of historical inter arrival times. The previous chart is an example for standard deviation of 200 ms. If the heartbeats arrive with less deviation the curve becomes steeper, i.e. it is possible to determine failure more quickly. The curve looks like this for a standard deviation of 100 ms.

To be able to survive sudden abnormalities, such as garbage collection pauses and transient network failures the failure detector is configured with a margin, akka.remote.watch-failure-detector.acceptable-heartbeat-pause. You may want to adjust the Remote Configuration of this depending on you environment. This is how the curve looks like for acceptable-heartbeat-pause configured to 3 seconds.

ByteBuffer based serialization

Artery introduces a new serialization mechanism which allows the ByteBufferSerializer to directly write into a shared java.nio.ByteBuffer instead of being forced to allocate and return an Array[Byte] for each serialized message. For high-throughput messaging this API change can yield significant performance benefits, so we recommend changing your serializers to use this new mechanism.

This new API also plays well with new versions of Google Protocol Buffers and other serialization libraries, which gained the ability to serialize directly into and from ByteBuffers.

As the new feature only changes how bytes are read and written, and the rest of the serializatio infrastructure remained the same, we recommend reading the Serialization documentation first.

Implementing an akka.serialization.ByteBufferSerializer works the same way as any other serializer,

interface ByteBufferSerializer {
  /**
   * Serializes the given object into the `ByteBuffer`.
   */
  void toBinary(Object o, ByteBuffer buf);
  
  /**
   * Produces an object from a `ByteBuffer`, with an optional type-hint;
   * the class should be loaded using ActorSystem.dynamicAccess.
   */
  void fromBinary(ByteBuffer buf, String manifest);
}

Implementing a serializer for Artery is therefore as simple as implementing this interface, and binding the serializer as usual (which is explained in Serialization).

Implementations should typically extend SerializerWithStringManifest and in addition to the ByteBuffer based toBinary and fromBinary methods also implement the array based toBinary and fromBinary methods. The array based methods will be used when ByteBuffer is not used, e.g. in Akka Persistence.

Note that the array based methods can be implemented by delegation like this:

import akka.serialization.ByteBufferSerializer;
import akka.serialization.SerializerWithStringManifest;

  class ExampleByteBufSerializer extends SerializerWithStringManifest
    implements ByteBufferSerializer {

    @Override
    public int identifier() {
      return 1337;
    }

    @Override
    public String manifest(Object o) {
      return "serialized-" + o.getClass().getSimpleName();
    }

    @Override
    public byte[] toBinary(Object o) {
      // in production code, aquire this from a BufferPool
      final ByteBuffer buf = ByteBuffer.allocate(256);

      toBinary(o, buf);
      buf.flip();
      final byte[] bytes = new byte[buf.remaining()];
      buf.get(bytes);
      return bytes;
    }

    @Override
    public Object fromBinary(byte[] bytes, String manifest) {
      return fromBinary(ByteBuffer.wrap(bytes), manifest);
    }

    @Override
    public void toBinary(Object o, ByteBuffer buf) {
      // Implement actual serialization here
    }

    @Override
    public Object fromBinary(ByteBuffer buf, String manifest) {
      // Implement actual deserialization here
      return null;
    }
  }

Disabling the Java Serializer

With Artery it is possible to completely disable Java Serialization for the entire Actor system.

Java serialization is known to be slow and prone to attacks of various kinds - it never was designed for high throughput messaging after all. However it is very convenient to use, thus it remained the default serialization mechanism that Akka used to serialize user messages as well as some of its internal messages in previous versions. Since the release of Artery, Akka internals do not rely on Java serialization anymore (exceptions to that being java.lang.Throwable and "remote deployment").

For user messages, the default serializer, implemented using Java serialization, remains available and enabled in Artery. We do however recommend to disable it entirely and utilise a proper serialization library instead in order effectively utilise the improved performance and ability for rolling deployments using Artery. Libraries that we recommend to use include, but are not limited to, Kryo by using the akka-kryo-serialization library or Google Protocol Buffers if you want more control over the schema evolution of your messages.

In order to completely disable Java Serialization in your Actor system you need to add the following configuration to your application.conf:

akka {
  actor {
    serialization-bindings {
      "java.io.Serializable" = none
    }
  }
}

Please note that this means that you will have to configure different serializers which will able to handle all of your remote messages. Please refer to the Serialization documentation as well as ByteBuffer based serialization to learn how to do this.

Dedicated subchannel for large messages

All the communication between user defined remote actors are isolated from the channel of Akka internal messages so a large user message cannot block an urgent system message. While this provides good isolation for Akka services, all user communications by default happen through a shared network connection (an Aeron stream). When some actors send large messages this can cause other messages to suffer higher latency as they need to wait until the full message has been transported on the shared channel (and hence, shared bottleneck). In these cases it is usually helpful to separate actors that have different QoS requirements: large messages vs. low latency.

Akka remoting provides a dedicated channel for large messages if configured. Since actor message ordering must not be violated the channel is actually dedicated for actors instead of messages, to ensure all of the messages arrive in send order. It is possible to assign actors on given paths to use this dedicated channel by using path patterns:

akka.remote.artery.large-message-destinations = [
   "/user/largeMessageActor",
   "/user/largeMessagesGroup/*",
   "/user/anotherGroup/*/largeMesssages",
   "/user/thirdGroup/**",
]

This means that all messages sent to the following actors will pass through the dedicated, large messages channel:

• /user/largeMessageActor
• /user/largeMessageActorGroup/actor1
• /user/largeMessageActorGroup/actor2
• /user/anotherGroup/actor1/largeMessages
• /user/anotherGroup/actor2/largeMessages
• /user/thirdGroup/actor3/
• /user/thirdGroup/actor4/actor5

Messages destined for actors not matching any of these patterns are sent using the default channel as before.

External, shared Aeron media driver

The Aeron transport is running in a so called media driver. By default, Akka starts the media driver embedded in the same JVM process as application. This is convenient and simplifies operational concerns by only having one process to start and monitor.

The media driver may use rather much CPU resources. If you run more than one Akka application JVM on the same machine it can therefore be wise to share the media driver by running it as a separate process.

The media driver has also different resource usage characteristics than a normal application and it can therefore be more efficient and stable to run the media driver as a separate process.

Given that Aeron jar files are in the classpath the standalone media driver can be started with:

java io.aeron.driver.MediaDriver

The needed classpath:

Agrona-0.5.4.jar:aeron-driver-1.0.1.jar:aeron-client-1.0.1.jar

You find those jar files on maven central, or you can create a package with your preferred build tool.

You can pass Aeron properties as command line -D system properties:

-Daeron.dir=/dev/shm/aeron

You can also define Aeron properties in a file:

java io.aeron.driver.MediaDriver config/aeron.properties

An example of such a properties file:

aeron.mtu.length=16384
aeron.socket.so_sndbuf=2097152
aeron.socket.so_rcvbuf=2097152
aeron.rcv.buffer.length=16384
aeron.rcv.initial.window.length=2097152
agrona.disable.bounds.checks=true

aeron.threading.mode=SHARED_NETWORK

# low latency settings
#aeron.threading.mode=DEDICATED
#aeron.sender.idle.strategy=org.agrona.concurrent.BusySpinIdleStrategy
#aeron.receiver.idle.strategy=org.agrona.concurrent.BusySpinIdleStrategy

# use same director in akka.remote.artery.advanced.aeron-dir config
# of the Akka application
aeron.dir=/dev/shm/aeron

Read more about the media driver in the Aeron documentation.

To use the external media driver from the Akka application you need to define the following two configuration properties:

akka.remote.artery.advanced {
  embedded-media-driver = off
  aeron-dir = /dev/shm/aeron
}

The aeron-dir must match the directory you started the media driver with, i.e. the aeron.dir property.

Several Akka applications can then be configured to use the same media driver by pointing to the same directory.

Note that if the media driver process is stopped the Akka applications that are using it will also be stopped.

Aeron Tuning

See Aeron documentation about Performance Testing.

Fine-tuning CPU usage latency tradeoff

Artery has been designed for low latency and as a result it can be CPU hungry when the system is mostly idle. This is not always desirable. It is possible to tune the tradeoff between CPU usage and latency with the following configuration:

# Values can be from 1 to 10, where 10 strongly prefers low latency # and 1 strongly prefers less CPU usage akka.remote.artery.advanced.idle-cpu-level = 1

By setting this value to a lower number, it tells Akka to do longer "sleeping" periods on its thread dedicated for spin-waiting and hence reducing CPU load when there is no immediate task to execute at the cost of a longer reaction time to an event when it actually happens. It is worth to be noted though that during a continuously high-throughput period this setting makes not much difference as the thread mostly has tasks to execute. This also means that under high throughput (but below maximum capacity) the system might have less latency than at low message rates.

Akka behind NAT or in a Docker container

In setups involving Network Address Translation (NAT), Load Balancers or Docker containers the hostname and port pair that Akka binds to will be different than the "logical" host name and port pair that is used to connect to the system from the outside. This requires special configuration that sets both the logical and the bind pairs for remoting.

akka {
  remote {
    artery {
      canonical.hostname = my.domain.com      # external (logical) hostname
      canonical.port = 8000                   # external (logical) port

      bind.hostname = local.address # internal (bind) hostname
      bind.port = 25520              # internal (bind) port
    }
 }
}