Riak_Core with Elixir : Part Three

This is the third post in the riak_core series.

1. Read part one here.
2. Read part two here.

In this post we will dwell deep into Vnodes in riak_core.

You can think of Vnodes as virtual nodes which potentially stores data and business logic.

In riak_core Vnodes servers as a basic unit of concurrency, fault-tolerance and replication and their life cycle is managed riak_core.

Most of the work involved in managing these Vnodes is done by riak_core, and all we need to do is to define a module which implements riak_core_vnode behaviour. If you are new to elixir or erlang, think of a behavior like an interface in that it declares callback functions that you must implement.

Required callback functions that we need to implement are :

init/1, handle_command/3, handle_coverage/4, handle_exit/3, handoff_starting/2, handoff_cancelled/1, handoff_finished/2, handle_handoff_command/3, handle_handoff_data/2, encode_handoff_item/2, is_empty/1, terminate/2, delete/1.

Since our app does not have any complex logic, we will restrict ourself only to implementation details of most important callbacks which are essential to understand the Vnode behaviour.

Defining a behaviour in elixir can be done by adding below line in our Vyuha.Vnode module.

@behaviour :riak_core_vnode

Life-Cycle callbacks

init/1 and terminate/2 are two callback which are called at the extreme end of Vnode life cycle.

init/1

If you think of riak_core_ring as circular array (of elixir/erlang process), than the space denoted by each index is a vnode. Index of this Vnode is called partition. So in order to initialize a Vnode we need to give it a partition number for which this Vnode is responsible for, This can be done by defining init/1 callback as shown in below line

def init([partition]) do
    {:ok, %{partition: partition}}
  end

The number of Vnodes in riak_core ring are decided at the time of installation and hence are fixed. Roughly if there are  n nodes and m being the total number of Vnodes than each server will have m/n nodes.

terminate/2

This callback is used to cleanup any resources held by the vnode. Here “reason” in arguments list is the reason why the given node was stopped.The State is the final state of the vnode and Result can be anything but will be ignored by the container.

def terminate(reason, state) do
    :ok
  end

For our simple app we return :ok.

handle_exit/3

When a process linked to the vnode dies this callback will be invoked with the Pid of the crashed process along with the Reason for the crash and the vnode's current State. At this point you have two choices.

1. The linked Pid is vital to the functioning of the vnode so you return {stop, NewState} to bring the vnode down as well.
2. The linked Pid isn't vital so you log a warning and return {noreply, NewState} to allow the vnode to continue execution.

def handle_exit(pid, reason, state) do
    {:noreply, state}
  end

Commands

This is the most important part of Vnode because this is the place where we define our business logic.

To add a command you add a new functional clause to that matches against the incoming request. Function signature of handle_command/3 is as below:

handle_command(Request, Sender, State) -> Result

where, Request can be anything (an atom or a tuple). The Sender is a representation of the client process but is typically used as an opaque value that you would use with a utility function such as riak_core_vnode:reply/2. The State is much like state in a gen_server and is there to track data across callback invocations

Result can take forms

1. {:reply, Reply, NewState} -> reply with a response and new state.
2. {:noreply, NewState} -> don’t reply with new state.

{:stop, reason, NewState} -> stop and terminate the vnode process and reply with a reason and new state.

Think of states as data, for example state could be a hashmap that we receive as a function parameter and every time we add a new entry in that hashmap, we pass it in the response as new state so that different callbacks can keep track of this hashmap.

we will go ahead with the first option and reply with :pong.

def handle_command(:ping, _sender, state) do
    Logger.warn("got a ping request!")
    {:reply, :pong, state}
  end

For our app these are the most important callback that we need to understand. For other callback we will give a default implementation.Our Vyuha.Vnode module should look something like this after implementing all the callbacks.

defmodule Vyuha.Vnode do
  require Logger
  @behaviour :riak_core_vnode
def start_vnode(partition) do
    :riak_core_vnode_master.get_vnode_pid(partition, __MODULE__)
  end
def init([partition]) do
    {:ok, %{partition: partition}}
  end
def handle_command(:ping, _sender, state) do
    Logger.warn("got a ping request!")
    {:reply, :pong, state}
  end
def handle_handoff_command(_fold_req, _sender, state) do
    {:noreply, state}
  end
def handoff_starting(_target_node, state) do
    {true, state}
  end
def handoff_cancelled(state) do
    {:ok, state}
  end
def handoff_finished(_target_node, state) do
    {:ok, state}
  end
def handle_handoff_data(data, state) do
    {:reply, :ok, state}
  end
def encode_handoff_item(object_name, object_value) do
    ""
  end
def is_empty(state) do
    {true, state}
  end
def delete(state) do
    {:ok, state}
  end
def handle_coverage(req, key_spaces, sender, state) do
    {:stop, :not_implemented, state}
  end
def handle_exit(pid, reason, state) do
    {:noreply, state}
  end
def terminate(reason, state) do
    :ok
  end
end

If you are really curious about other callbacks and how to implement them than read this awesome post.