As we said before, a node is an executing Erlang system, and a node is said to be alive if it can communicate with other nodes; this is another way of saying that the node is named, and so can take part in communication.

The function erlang:is_alive() will test whether the local runtime system is alive, and as you can see from the following example, it is possible to change the live status of a running runtime system using functions from the net_kernel module, as well as finding out the name of the current node using the node/0 BIF. Try it out:

1> erlang:is_alive().
false
2> net_kernel:start([foo]).
{ok,<0.33.0>}
([email protected])3> erlang:is_alive().
true
([email protected])4> node().
'[email protected]'
([email protected])5> net_kernel:stop().
ok
6> erlang:is_alive().
false

Each live node has to be named: these names must be unique on that host, but can be duplicated across different hosts. The name/host pair, called the identifier of the node, is used to uniquely identify the node in the network.

Names take two forms. You already saw the first one; the second one is new:

Nodes with long names can communicate only with other nodes with long names; similarly for nodes with short names.

For two nodes to communicate, not only must both of them be alive, but also they must share some information contained in an atom called the secret cookie. Each node has a single cookie value at any time, and nodes sharing the same value can communicate.

Each node can be started with an explicit cookie value, as in the following:

erl -sname foo -setcookie blah

If no value is set on launch, the Erlang runtime system will pick up the value stored in the file .erlang.cookie. If the file does not exist, it will be created in the home directory of the user’s account. A randomly generated secret cookie value will be stored in it. As a result, nodes created on the same user account will share the same cookie value by default. If you have been experimenting with distributed Erlang without setting a cookie, look for the .erlang.cookie file. You can edit it to whatever value you want.

To show secret cookies and distribution in action, we have repeated the example from Figure 11-2 with nodes running on separate host computers on the same network; this is shown in Figure 11-3.

Note in Figure 11-3 how the two nodes are explicitly started with the same cookie value; this value will override any values contained in an .erlang.cookie file on either host.

Distributing the Erlang code: A warning

Your first attempt to distribute the Erlang code may fail: why? The following call:

(foo@STC)1> spawn('bar@FCC', dist, t, [self()]).

is on the node foo at host STC, but the spawned code is executed on FCC. The module dist.erl will not be executable on FCC unless it is there, as it will not be transferred from STC to FCC automatically.

Figure 11-3. Example with nodes on different hosts

Moreover, the call will still fail unless there is a compiled version of the code in the code search path on the remote Erlang node on host FCC.

Erlang Distribution and Security

When a distributed node on your machine gets connected to a remote node through a shared cookie, the owner of the remote node gains the same user access rights as the account on which your local Erlang node is running.

This might be acceptable in a closed telecom system or a banking system running behind a firewall, but if you are running the node on your personal account, anyone connected to it would be able to read and delete files, execute commands, and hijack your machine. Although the caller of the following function:

spawn(YourNode, os, cmd, ["rm -rf *"])

might find it funny, you might not enjoy the trail of peace and tranquility the call leaves behind.^[28]

As a result, you should never publish your Erlang node name and cookie to anyone, unless you’ve adapted the net kernel to cater for security issues or you explicitly trust the person not to do anything malicious.

The most elementary communication is for one node to test whether it can communicate with another, a process informally known as pinging the node (see Figure 11-4).

Figure 11-4. Pinging a node

Figure 11-4 shows two nodes initialized with different cookie values, which will prevent their communication. The foo node attempts to communicate using net_adm:ping/1: the pang response shows that this fails. The pinged node also gives an error report to signal the connection attempt, by way of warning that a potential security problem has occurred. After changing the cookie for foo to cake, the ping is successful, indicated by the pong result; such a successful attempt is not signaled at the bar node.

Next, we’ll look at an example in which a call to spawn/4 registers a process on a remote node, and then communicates with it; this is illustrated in Figure 11-5.

Figure 11-5. Communication with a registered process on another node

The first command in the foo node spawns the dist:s/0 process. The effect of this is to register the loop loop/0 under the name server. The effect of this loop is to repeatedly receive messages of the form {M, Pid} and to return the message M to the Pid. In the second command at foo, the message hi is sent together with the Pid of the shell running at foo. This is received by server, and the hi message is returned to foo; you can see this in the inbox of foo as a result of the flush() of the inbox.

Distributed Erlang nodes are able to communicate with each other, provided that they share the same cookie information, but you haven’t seen how the connections between the nodes are set up. That is because the Erlang runtime system sets up the connection automatically to a node when it is first referred to. This might be through the net_adm:ping/1 call or by sending a message to a registered process on it. Information about nodes is, by default, shared between connected nodes so that if A knows about B, and B about C, then A will also find out about C.

Each node has a cookie value at any one time. Security in distributed Erlang is based on sharing cookie information: what happens when cookie values are changed? The following interaction shows this in action:

(foo@STC)1> net_adm:ping('bar@STC').
pang
(foo@STC)2> erlang:set_cookie(node(),cake).
true
(foo@STC)3> net_adm:ping('bar@STC').
pong
(foo@STC)4> erlang:set_cookie(node(),fish).
true
(foo@STC)5> net_adm:ping('bar@STC').
pong

The nodes foo and bar initially have different cookies, hence the negative reply to the first command. In command 2, the cookie of foo is set to that of bar, and so in command 3, the ping is successful, because a connection can be established. In command 4, the cookie is changed back to its original value, but command 5 shows that the two nodes remain connected despite the fact that the two nodes now have different cookies.

This “inclusive” model of connection may not be what is required, and using the facilities of the net_kernel module it is possible to control connections by hand. It is also possible to use the erl command with the flag -connect_all false to avoid nodes from globally connecting to each other.

The net_kernel process at each node coordinates operations at a distributed Erlang node. BIFs such as spawn/4 are converted by the net_kernel to messages that are sent to the net_kernel on the remote node. The net_kernel process also handles authentication by cookies. Because the net_kernel is simply another Erlang process, it is possible for a user to modify it to provide a different behavior, such as changing the authentication scheme or not allowing processes from another node to spawn processes.

Even the most security-unconscious readers will have realized that basing your security on secret cookies alone is not very reliable. As telecom clusters tend to run behind firewalls, enhancing security in its distribution model has never been an issue. In the early days, cookies were in fact sent across the network unencrypted!

Considering the low level of security in distributed Erlang, how can you build a secure distributed system in Erlang? There are two answers to this question:

You can enhance security by writing your own net_kernel process, giving the process whatever behavior and level of security you might require.

erl -sname foo -hidden

(alpha@STC)1> net_kernel:connect('beta@STC').
true
(alpha@STC)2> net_kernel:connect('gamma@STC').
true
(alpha@STC)3> nodes().
['beta@STC']
(alpha@STC)4> nodes(hidden).
['gamma@STC']
(alpha@STC)5> nodes(connected).
['beta@STC', 'gamma@STC']

UNIXSHELL> erl -sname beta
Erlang (BEAM) emulator version 5.5
Eshell V5.5 (abort with ^G)
(beta@STC)1> nodes().
['alpha@STC']
(beta@STC)2> nodes(connected).
['beta@STC']

UNIXSHELL> erl -sname gamma -hidden
Erlang (BEAM) emulator version 5.5
Eshell V5.5 (abort with ^G)
(gamma@STC)1> nodes().
[]
(gamma@STC)2> nodes(hidden).
['alpha@STC']

The classic construct in distributed computing is the remote procedure call (RPC) in which a local call to a procedure is replaced by a call to the same procedure running on a remote node. RPC is simple to implement in Erlang, and moreover, the Erlang implementation avoids many of the pitfalls of RPC in other languages.

In the basic Erlang implementation of RPC, the following (local) function call,

Val = fac(N)

is replaced by a message to send and receive, as shown here and illustrated in Figure 11-6:

remote_call(Message, Node) ->
  {facserver, Node} ! {self(), Message},
  receive
    {ok, Res} ->
      Res
  end.

Figure 11-6. Remote procedure call

In the following code, the facserver process is on the bar@STC node and runs in the facLoop/0 loop function:

server() ->
  register(facserver,self()),
  facLoop().

facLoop() ->
  receive
    {Pid, N} ->
      Pid ! {ok, fac(N)}
  end,
  facLoop().

The main difference between a local and a remote call is the fact that a remote node may go down. You can deal with this in a number of different ways. For example, you can add a timeout in the client code:

remote_call(Message, Node) ->
  {facserver, Node} ! {self(), Message},
  receive
    {ok, Res} ->
      Res
    after 1000 ->
      {error, timeout}
  end.

If no reply is received within one second, the tuple {error, timeout} is returned. You have to be careful when using timeouts, as the message might still be received after the timeout and stored in the process mailbox. The remote server might be extremely busy or the network may be highly congested. If you do not flush the message, the next time remove_call/2 is invoked and a new request to the factorial server is sent, you’ll end up retrieving the first message in the queue containing the reply from the previous call.

Alternatively, it is possible to link to the server process, so if that fails, the client process will also fail. You can do this by launching the server process using spawn_link/4 rather than spawn/4:

setup() ->
  process_flag(trap_exit, true),
  spawn_link('bar@STC',myrpc,server,[]).

If the remote process terminates, you will receive the usual 'EXIT' signal. If the network connection between the two nodes goes down, the network kernel sends an 'EXIT' signal with reason noconnection.

Finally, it is possible to monitor whether a node is alive; the monitor_node(Node,Bool) BIF will switch this on/off for Node according to the Boolean flag Bool, as in the following code. When monitoring is active, the message {nodedown, Node} will be sent to the monitoring process:

remote_call(Message, Node) ->
  monitor_node(Node,true),
  {facserver, Node} ! {self(), Message},
    receive
      {ok, Res} ->
        monitor_node(Node,false),
        Res;
      {nodedown, Node} ->
        {error, node_down}
    end.

Don’t forget to demonitor your node when you are done, because calling monitor_node(Node, true) will generate a nodedown message for each time the BIF was called.

The rpc library module provides implementations of services that are similar to remote procedure calls, as well as facilities for broadcast and parallel evaluation of RPC calls. The most commonly used function is:

rpc:call(Node, Module, Function, Arguments)

which executes the function on the remote node. The Module must be in the code search path on the remote node, and the nodes must either be connected or share the cookie. The result is the return value of the call, or, upon failure, {badrpc, Reason}.

If your applications are going to be distributed, spend some time reading through the manual pages of the rpc module and become familiar with it. You will find help functions for synchronous, asynchronous, and blocking calls. There are also calls to broadcast calls, both synchronously and asynchronously to a pool of nodes. You never know when you are going to need these functions, so reviewing them now will avoid your having to reinvent the wheel at a later date.

A number of key modules support distributed programming in Erlang. Some we have already covered, while others are new:

When running distributed Erlang nodes behind firewalls, you need to open the port on which epmd listens. By default, this is port 4369, but you can change it to whatever port you please, as long as it is consistent in your node cluster. You also need to open the ports that the individual nodes use to connect to each other. You can specify the node range by running the following commands:

application:set_env(kernel, inet_dist_listen_min, 9100)
application:set_env(kernel, inet_dist_listen_max, 9105)

These commands force Erlang to use ports from 9100 to 9105 for distribution. You can replace these values with whatever range you want.

Design a distributed version of an associative store in which values are associated with tags. It is possible to store a tag/value pair, and to look up the value(s) associated with a tag. One example for this is an address book for email, in which email addresses (values) are associated with nicknames (tags).

Replicate the store across two nodes on the same host, send lookups to one of the nodes (chosen either at random or alternately), and send updates to both.

Reimplement your system with the store nodes on other hosts (from each other and from the frontend). What do you have to be careful about when you do this?

How could you reimplement the system to include three or four store nodes?

Design a system to test your answer to this exercise. This should generate random store and lookup requests.

Design a system to monitor the behavior of your distributed store systems under test conditions. This system—which could be another node in the overall system—should log throughput and load-balancing information. How does the system behave when it becomes overloaded?