View on GitHub

Derecho

Distributed systems toolkit for RDMA

A Brief User Guide

Contents

Installation

Derecho is a library that helps you build replicated, fault-tolerant services in a datacenter with RDMA networking. Here’s how to start using it in your projects.

Prerequisites

Getting Started

To download the project, run

git clone https://github.com/Derecho-Project/derecho.git

Once cloning is complete, to build the code, cd into the derecho directory and run:

This will place the binaries and libraries in the sub-directories of Release. The other build type is Debug. If you need to build the Debug version, replace Release by Debug in the above instructions. We explicitly disable in-source build, so running cmake . in derecho will not work.

Once the project is built, install it by running:

By default, Derecho will be installed into /usr/local/. Please make sure you have sudo privileges to write to system directories.

Successful installation will set up the followings in $DESTDIR:

To uninstall, run:

To build your own derecho executable, simple run:

To use Derecho in your code, you simply need to

The configuration file consists of three sections: DERECHO, RDMA, and PERS. The DERECHO section includes core configuration options for a Derecho instance, which every application will need to customize. The RDMA section includes options for RDMA hardware specifications. The PERS section allows you to customize the persistent layer’s behavior.

Configuring Core Derecho

Applications need to tell the Derecho library which node is the initial leader with the options leader_ip and leader_gms_port. Each node then specifies its own ID (local_id) and the IP address and ports it will use for Derecho component services (local_ip, gms_port, state_transfer_port, sst_port, and rdmc_port). Also, if using external clients, applications need to specify the ports serving external clients (leader_external_port and external_port);

The other important parameters are the message sizes. Since Derecho pre-allocates buffers for RDMA communication, each application should decide on an optimal buffer size based on the amount of data it expects to send at once. If the buffer size is much larger than the messages an application actually sends, Derecho will pin a lot of memory and leave it underutilized. If the buffer size is smaller than the application’s actual message size, it will have to split messages into segments before sending them, causing unnecessary overhead.

Three message-size options control the memory footprint and performance of Derecho. In all cases, larger values will increase the memory (DRAM) footprint of the application, and it is fairly easy to end up with a huge memory size if you just pick giant values. The defaults keep the memory size smaller, but can reduce performance if an application is sending high rates of larger messages.

The options are named max_payload_size, max_smc_payload_size, block_size, max_p2p_request_payload_size, and max_p2p_reply_payload_size.

No message bigger than max_payload_size will be sent by Derecho multicast(Replicated<>::send()). No message bigger than max_p2p_request_payload_size will be sent by Derecho p2p send(Replicated<>::p2p_send() or ExternalClientCaller<>::p2p_send()). No reply bigger than max_p2p_reply_payload_size will be sent to carry the return values any multicast or p2p send.

To understand the other two options, it helps to remember that internally, Derecho makes use of two sub-protocols when it transmits your data. One sub-protocol is optimized for small messages, and is called SMC. Messages equal to or smaller than max_smc_payload_size will be sent using SMC. Normally max_smc_payload_size is set to a small value, like 1K, but we have tested with values up to 10K. This limit should not be made much larger: performance will suffer and memory would bloat.

Larger messages are sent via RDMC, our big object protocol. These will be automatically broken into chunks. Each chunk will be of size block_size. The block_size value we tend to favor in our tests is 1MB, but we have run experiments with values as large as 100MB. If you plan to send huge objects, like 100MB or even multi-gigabyte images, consider a larger block size: it pays off at that scale. If you expect that huge objects would be rare, use a value like 1MB.

More information about Derecho parameter setting can be found in the comments in the default configuration file. You may want to read about window_size, timeout_ms, and rdmc_send_algorithm.

Configuring RDMA Devices

The most important configuration entries in this section are provider and domain. The provider option specifies the type of RDMA device (i.e. a class of hardware) and the domain option specifies the device (i.e. a specific NIC or network interface). This Libfabric document explains the details of those concepts.

The tx_depth and rx_depth configure the maximum of pending requests that can be waiting for acknowledgement before communication blocks. Those numbers can be different from one device to another. We recommend setting them as large as possible.

Here are some sample configurations showing how Derecho might be configured for two common types of hardware.

Configuration 1: run Derecho over TCP/IP with Ethernet interface ‘eth0’:

...
[RDMA]
provider = sockets
domain = eth0
tx_depth = 256
rx_depth = 256
...

Configuration 2: run Derecho over verbs RDMA with RDMA device ‘mlx5_0’:

...
[RDMA]
provider = verbs
domain = mlx5_0
tx_depth = 4096
rx_depth = 4096
...

Configuring Persistent Behavior

The application can specify the location for persistent state in the file system with file_path, which defaults to the .plog folder in the working directory. ramdisk_path controls the location of states for Volatile<T>, which defaults to tmpfs (ramdisk). reset controls weather to clean up the persisted state when a Derecho service shuts down. We default this to true. Please set reset to false for normal use of Persistent<T>.

Specify Configuration with Command Line Arguments

We also allow applications to specify configuration options on the command line. Any command line configuration options override the equivalent option in configuration file. To use this feature while still accepting application-specific command-line arguments, we suggest using the following code:

#define NUM_OF_APP_ARGS () // specify the number of application arguments.
int main(int argc, char* argv[]) {
    if((argc < (NUM_OF_APP_ARGS+1)) ||
       ((argc > (NUM_OF_APP_ARGS+1)) && strcmp("--", argv[argc - NUM_OF_APP_ARGS - 1]))) {
        cout << "Invalid command line arguments." << endl;
        cout << "USAGE:" << argv[0] << "[ derecho-config-list -- ] application-argument-list" << endl;
        return -1;
    }
    Conf::initialize(argc, argv); // pick up configurations in the command line list
    // pick up the application argument list and continue ...
    ...
}

Then, call the application as follows, assuming the application’s name is app:

$ app --DERECHO/local_id=0 --PERS/reset=false -- <application-argument-list>

Please refer to the bandwidth_test application for more details.

Setup and Testing

There are some sample programs in the folder applications/demos that can be run to test the installation. In addition, there are some performance tests in the folder applications/tests/performance_tests that you may want to use to measure the performance Derecho achieves on your system. To be able to run the tests, you need a minimum of two machines connected by RDMA. The RDMA devices on the machines should be active. In addition, you need to run the following commands to install and load the required kernel modules for using RDMA hardware:

sudo apt-get install rdmacm-utils ibutils libmlx4-1 infiniband-diags libmthca-dev opensm ibverbs-utils libibverbs1 libibcm1 libibcommon1
sudo modprobe -a rdma_cm ib_uverbs ib_umad ib_ipoib mlx4_ib iw_cxgb3 iw_cxgb4 iw_nes iw_c2 ib_mthca

Depending on your system, some of the modules might not load which is fine.

RDMA requires memory pinning of memory regions shared with other nodes. There’s a limit on the maximum amount of memory a process can pin, typically 64 KB, which Derecho easily exceeds. Therefore, you need to set this to unlimited. To do so, append the following lines to /etc/security/limits.conf:

* [username] hard memlock unlimited
* [username] soft memlock unlimited

Derecho maintains many TCP connections as well as disk files for large scale setups. We recommend raise the limit of maximum number of open files by appending the following lines to /etc/security/limits.conf:

* hard nofile 10240
* soft nofile 10240

where [username] is your linux username. A * in place of the username will set this limit to unlimited for all users. Log out and back in again for the limits to reapply. You can test this by verifying that ulimit -l outputs unlimited in bash.

The persistence layer of Derecho stores durable logs of updates in memory-mapped files. Linux also limits the size of memory-mapped files to a small size that Derecho usually exceeds, so you will need to set the system parameter vm.overcommit_memory to 1 for persistence to work. To do this, run the command

sysctl -w vm.overcommit_memory = 1

A simple test to see if your setup is working is to run the test bandwidth_test from applications/tests/performance_tests. To run it, go to two of your machines (nodes), cd to Release/src/applications/tests/performance_tests and run ./bandwidth_test 2 0 100000 0 on both. As a confirmation that the experiment finished successfully, the first node will write a log of the result in the file data_derecho_bw, which will be something along the lines of 2 0 10240 300 100000 0 5.07607. Full experiment details including explanation of the arguments, results and methodology is explained in the source documentation for this program.

Using Derecho

The file simple_replicated_objects.cpp within applications/demos shows a complete working example of a program that sets up and uses a Derecho group with several Replicated Objects. You can read through that file if you prefer to learn by example, or read on for an explanation of how to use various features of Derecho.

Replicated Objects

One of the core building blocks of Derecho is the concept of a Replicated Object. This provides a simple way for you to define state that is replicated among several machines and a set of RPC functions that operate on that state. A Replicated Object is any class that (1) is serializable with the mutils-serialization framework and (2) implements a static method called register_functions().

The mutils-serialization library should have more documentation on making objects serializable, but the most straightforward way is to inherit mutils::ByteRepresentable, use the macro DEFAULT_SERIALIZATION_SUPPORT, and write an element-by-element constructor. The register_functions() method is how your class specifies to Derecho which of its methods should be converted to RPC functions and what their numeric “function tags” should be. It should return a std::tuple containing a pointer to each RPC-callable method, wrapped in the template functions derecho::rpc::tag_p2p or derecho::rpc::tag_ordered, depending on how the method will be invoked. Methods wrapped in tag_p2p can be called by peer-to-peer RPC messages and must not modify the state of the Replicated Object, while methods wrapped in tag_ordered can be called by ordered-multicast RPC messages and can modify the state of the object. We have provided a default implementation of this function that is generated with the macros REGISTER_RPC_FUNCTIONS, P2P_TARGETS, and ORDERED_TARGETS; the P2P_TARGETS and ORDERED_TARGETS macros tag their arguments as P2P and ordered-callable RPC methods, respectively, while the REGISTER_RPC_FUNCTIONS macro combines the results of these macros. Here is an example of a Replicated Object declaration that uses the default implementation macros:

class Cache : public mutils::ByteRepresentable {
    std::map<std::string, std::string> cache_map;

public:
    void put(const std::string& key, const std::string& value);
    std::string get(const std::string& key) const;
    bool contains(const std::string& key) const;
    bool invalidate(const std::string& key);
    Cache() : cache_map() {}
    Cache(const std::map<std::string, std::string>& cache_map) : cache_map(cache_map) {}
    DEFAULT_SERIALIZATION_SUPPORT(Cache, cache_map);
    REGISTER_RPC_FUNCTIONS(Cache,
                           ORDERED_TARGETS(put, invalidate),
                           P2P_TARGETS(get, contains));
};

This object has one field, cache_map, so the DEFAULT_SERIALIZATION_SUPPORT macro is called with the name of the class and the name of this field. The second constructor, which initializes the field from a parameter of the same type, is required for serialization support. The object has two read-only RPC methods that should be invoked by peer-to-peer messages, get and contains, so these method names are passed to the P2P_TARGETS macro; similarly, it has two read-write RPC methods that should be invoked by ordered multicasts, put and invalidate, so these method names are passed to the ORDERED_TARGETS macro. The numeric function tags generated by REGISTER_RPC_FUNCTIONS can be re-generated with the macro RPC_NAME, so these functions can later be called by using the tags RPC_NAME(put), RPC_NAME(get) RPC_NAME(contains), and RPC_NAME(invalidate).

Groups and Subgroups

Derecho organizes nodes (machines or processes in a system) into Groups, which can then be divided into subgroups and shards. Any member of a Group can communicate with any other member, and all run the same group-management service that handles failures and accepts new members. Subgroups, which are any subset of the nodes in a Group, correspond to Replicated Objects; each subgroup replicates the state of a Replicated Object and any member of the subgroup can handle RPC calls on that object. Shards are disjoint subsets of a subgroup that each maintain their own state, so one subgroup can replicate multiple instances of the same type of Replicated Object. A Group must be statically configured with the types of Replicated Objects it can support, but the number of subgroups and their exact membership can change at runtime according to functions that you provide.

Note that more than one subgroup can use the same type of Replicated Object, so there can be multiple independent instances of a Replicated Object in a Group even if those subgroups are not sharded. A subgroup is usually identified by the type of Replicated Object it implements and an integral index number specifying which subgroup of that type it is.

To start using Derecho, a process must either start or join a Group by constructing an instance of derecho::Group, which then provides the interface for interacting with other nodes in the Group. (The process will start a new Group if it is configured as the Group leader, otherwise it joins the existing group by contacting the configured leader). A derecho::Group expects a set of variadic template parameters representing the types of Replicated Objects that it can support in its subgroups. For example, this declaration is a pointer to a Group object that can have subgroups of type LoadBalancer, Cache, and Storage:

std::unique_ptr<derecho::Group<LoadBalancer, Cache, Storage>> group;

Defining Subgroup Membership

In order to start or join a Group, all members (including processes that join later) must define a function that provides the membership (as a subset of the current View) for each subgroup. The membership function’s input is the list of Replicated Object types, the current View, and the previous View if there was one. Its return type is a std::map mapping each Replicated Object type to a vector representing all the subgroups of that type (since there can be more than one subgroup that implements the same Replicated Object type). Each entry in this vector is another vector, whose size indicates the number of shards the subgroup should be divided into, and whose entries are SubViews describing the membership of each shard. For example, if the membership function’s return value is named members, then members[std::type_index(typeid(Cache))][0][2] is a SubView identifying the members of the third shard of the first subgroup of type “Cache.”

Derecho provides a default subgroup membership function that automatically assigns nodes from the Group into disjoint subgroups and shards, given a policy that describes the desired number of nodes in each subgroup/shard. It assigns nodes in ascending rank order, and leaves any “extra” nodes (not needed to fully populate all subgroups) at the end (highest rank) of the membership list. During a View change, this function attempts to preserve the correct number of nodes in each shard without re-assigning any nodes to new roles. It does this by copying the subgroup membership from the previous View as much as possible, and assigning idle nodes from the end of the Group’s membership list to replace failed members of subgroups.

There are several helper functions in subgroup_functions.hpp that construct AllocationPolicy objects for different scenarios, to make it easier to set up the default subgroup membership function. Here is an example of how the default membership function could be configured for two types of Replicated Objects using these functions:

derecho::SubgroupInfo subgroup_function {derecho::DefaultSubgroupAllocator({
    {std::type_index(typeid(Foo)), derecho::one_subgroup_policy(derecho::even_sharding_policy(2,3))},
    {std::type_index(typeid(Bar)), derecho::identical_subgroups_policy(
            2, derecho::even_sharding_policy(1,3))}
})};

Based on the policies constructed for the constructor argument of DefaultSubgroupAllocator, the function will create one subgroup of type Foo, with two shards of 3 members each. Next, it will create two subgroups of type Bar, each of which has only one shard of size 3. Note that the order in which subgroups are allocated is the order in which their Replicated Object types are listed in the Group’s template parameters, so this instance of the default subgroup allocator will assign the first 6 nodes to the Foo subgroup and the second 6 nodes to the Bar subgroups the first time it runs.

More advanced users may, of course, want to define their own subgroup membership functions. The demo program overlapping_replicated_objects.cpp shows a relatively simple example of a user-defined membership function. In this program, the SubgroupInfo contains a C++ lambda function that implements the shard_view_generator_t type signature and handles subgroup assignment for Replicated Objects of type Foo, Bar, and Cache:

[](const std::vector<std::type_index>& subgroup_type_order,
   const std::unique_ptr<derecho::View>& prev_view, derecho::View& curr_view) {
    derecho::subgroup_allocation_map_t subgroup_allocation;
    for(const auto& subgroup_type : subgroup_type_order) {
        derecho::subgroup_shard_layout_t subgroup_layout(1);
        if(subgroup_type == std::type_index(typeid(Foo)) || subgroup_type == std::type_index(typeid(Bar))) {
            // must have at least 3 nodes in the top-level group
            if(curr_view.num_members < 3) {
                throw derecho::subgroup_provisioning_exception();
            }
            std::vector<node_id_t> first_3_nodes(&curr_view.members[0], &curr_view.members[0] + 3);
            //Put the desired SubView at subgroup_layout[0][0] since there's one subgroup with one shard
            subgroup_layout[0].emplace_back(curr_view.make_subview(first_3_nodes));
            //Advance next_unassigned_rank by 3, unless it was already beyond 3, since we assigned the first 3 nodes
            curr_view.next_unassigned_rank = std::max(curr_view.next_unassigned_rank, 3);
        } else { //subgroup_type == std::type_index(typeid(Cache))
            // must have at least 6 nodes in the top-level group
            if(curr_view.num_members < 6) {
                throw derecho::subgroup_provisioning_exception();
            }
            std::vector<node_id_t> next_3_nodes(&curr_view.members[3], &curr_view.members[3] + 3);
            subgroup_layout[0].emplace_back(curr_view.make_subview(next_3_nodes));
            curr_view.next_unassigned_rank += 3;
        }
        subgroup_allocation.emplace(subgroup_type, std::move(subgroup_layout));
    }
    return subgroup_allocation;
};

For all three types of Replicated Object, the function creates one subgroup and one shard. For the Foo and Bar subgroups, it assigns first three nodes in the current View’s members list (thus, these subgroups are co-resident on the same three nodes), while for the Cache subgroup it assigns nodes 3 to 6 on the current View’s members list. Note that if there are not enough members in the current view to assign 3 nodes to each subgroup, the function throws derecho::subgroup_provisioning_exception. This is how subgroup membership functions indicate to the view management logic that a view has suffered too many failures to continue executing (it is “inadequately provisioned”) and must wait for more members to join before accepting any more state updates.

Constructing a Group

Although the subgroup allocation function is the most important part of constructing a derecho::Group, it requires a few additional parameters.

Invoking RPC Functions

Once a process has joined a Group and one or more subgroups, it can invoke RPC functions on any of the Replicated Objects in the Group. The options a process has for invoking RPC functions depend on its membership status:

Ordered sends are invoked through the Replicated interface, whose template parameter is the type of the Replicated Object it communicates with. You can obtain a Replicated by using Group’s get_subgroup method, which uses a template parameter to specify the type of the Replicated Object and an integer argument to specify which subgroup of that type (remember that more than one subgroup can implement the same type of Replicated Object). For example, this code retrieves the Replicated object corresponding to the second subgroup of type Cache:

Replicated<Cache>& cache_rpc_handle = group->get_subgroup<Cache>(1);

The ordered_send method uses its template parameter, which is an integral “function tag,” to specify which RPC function it will invoke; if you are using the REGISTER_RPC_FUNCTIONS macro, the function tag will be the integer generated by the RPC_NAME macro applied to the name of the function. Its arguments are the arguments that will be passed to the RPC function call, and it returns an instance of derecho::rpc::QueryResults with a template parameter equal to the return type of the RPC function. Using the Cache example from earlier, this is what RPC calls to the “put” and “contains” functions would look like:

cache_rpc_handle.ordered_send<RPC_NAME(put)>("Foo", "Bar");
derecho::rpc::QueryResults<bool> results = cache_rpc_handle.ordered_send<RPC_NAME(contains)>("Foo");

P2P (peer-to-peer) sends are invoked through the ExternalCaller interface, which is exactly like the Replicated interface except that it only provides the p2p_send function. ExternalCaller objects are provided through the get_nonmember_subgroup method of Group, which works exactly like get_subgroup (except for the assumption that the caller is not a member of the requested subgroup). For example, this is how a process that is not a member of the second Cache-type subgroup would get an ExternalCaller to that subgroup:

ExternalCaller<Cache>& p2p_cache_handle = group->get_nonmember_subgroup<Cache>(1);

When invoking a P2P send, the caller must specify, as the first argument, the ID of the node to communicate with. The caller must ensure that this node is actually a member of the subgroup that the ExternalCaller targets (though it can be in any shard of that subgroup). Nodes can find out the current membership of a subgroup by calling the get_subgroup_members method on the Group, which uses the same template parameter and argument as get_subgroup to select a subgroup by type and index. For example, assuming Cache subgroups are not sharded, this is how a non-member process could make a call to get, targeting the first node in the second subgroup of type Cache:

std::vector<node_id_t> cache_members = group.get_subgroup_members<Cache>(1)[0];
derecho::rpc::QueryResults<std::string> results = p2p_cache_handle.p2p_send<RPC_NAME(get)>(cache_members[0], "Foo");

Using QueryResults objects

The result of an ordered send is a slightly complex object, because it must contain a std::future for each member of the subgroup, but the membership of the subgroup might change during the query invocation. Thus, a QueryResults object is actually itself a future, which is fulfilled with a map from node IDs to futures as soon as Derecho can guarantee that the query will be delivered in a particular View. (The node IDs in the map are the members of the subgroup in that View). Each std::future in the map will be fulfilled with either the response from that node or a node_removed_from_group_exception, if a View change occurred after the query was delivered but before that node had a chance to respond.

By the time the caller sees this exception, the view will have been updated. Thus a caller that wishes to reissue a request could do so immediately after the exception is caught: it can already look up the new membership, select a new target, and send a new request. On the other hand, notice that Derecho provides no indication of whether the target that failed did so before or after the original request was received. Thus if your target might have taken some action (like issuing an update request), you may have to include application-layer logic to make sure your reissued request won’t be performed twice if the initial request actually got past the update step, and the failure occurred later. A simple way to do this is to make your requests idempotent, for example by including a request-id and an “this is a retry” flag, and if the flag is true, having the group member check to see if that request-id has already been performed.

As an example, this code waits for the responses from each node and combines them to ensure that all replicas agree on an item’s presence in the cache:

derecho::rpc::QueryResults<bool> results = cache_rpc_handle.ordered_send<RPC_NAME(contains)>("Stuff");
bool contains_accum = true;
for(auto& reply_pair : results.get()) {
    bool contains_result = reply_pair.second.get();
    contains_accum = contains_accum && contains_result;
}

Note that the type of reply_pair is std::pair<derecho::node_id_t, std::future<bool>>, which is why a node’s response is accessed by writing reply_pair.second.get().

Tracking Updates with Version Vectors

Derecho allows tracking data update history with a version vector in memory or persistent storage. A new class template is introduced for this purpose: Persistent<T,ST>. In a Persistent instance, data is managed in an in-memory object of type T (we call it the “current object”) along with a log in a datastore specified by storage type ST. The log can be indexed using a version number, an index, or a timestamp. A version number is a 64-bit integer attached to each version; it is managed by the Derecho SST and guaranteed to be monotonic. A log is also an array of versions accessible using zero-based indices. Each log entry also has an attached timestamp (microseconds) indicating when this update happened according to the local real-time clock. To enable this feature, we need to manage the data in a serializable object T, and define a member of type Persistent<T> in the Replicated Object in a relevant group. Persistent_typed_subgroup_test.cpp gives an example.

/**
 * Example for replicated object with Persistent<T>
 */
class PFoo : public mutils::ByteRepresentable {
    Persistent<int> pint;
public:
    virtual ~PFoo() noexcept (true) {}
    int read_state() const {
        return *pint;
    }
    bool change_state(int new_int) {
         if(new_int == *pint) {
           return false;
         }

         *pint = new_int;
         return true;
    }

    // constructor with PersistentRegistry
    PFoo(PersistentRegistry * pr) : pint(nullptr,pr) {}
    PFoo(Persistent<int> & init_pint) : pint(std::move(init_pint)) {}
    DEFAULT_SERIALIZATION_SUPPORT(PFoo, pint);
    REGISTER_RPC_FUNCTIONS(PFoo, P2P_TARGETS(read_state), ORDERED_TARGETS(change_state));
};

For simplicity, the versioned type is int in this example. You set it up in the same way as a non-versioned member of a replicated object, except that you need to pass the PersistentRegistry from the constructor of the replicated object to the constructor of the Persistent<T>. Derecho uses PersistentRegistry to keep track of all the Persistent<T> objects in a single Replicated Object so that it can create versions on updates. The Persistent constructor registers itself in the registry.

By default, the Persistent<T> stores its log in the file-system (in a folder called .plog in the current directory). Applications can specify memory as the storage location by setting the second template parameter: Persistent<T,ST_MEM> (or Volatile<T> as syntactic sugar). We are working on more storage types including NVM.

Once the version vector is set up with Derecho, the application can query the value with the get() APIs in Persistent<T>. In persistent_temporal_query_test.cpp, a temporal query example is illustrated.