Rolling your own MOM, or how I did it - Project setup
Tags: cpp, seastar, reactor, asynchronous
Published: 2019-07-18 14:28:29

---
title: Rolling your own MOM, or how I did it - Project setup
series: MOM Series
published: true
description: 
tags: cpp, seastar, reactor, asynchronous
---


#Introduction
If you were bored reading the previous article, sorry about that, yes it was boring but unfortunately it's required. However, from now on, We get to the concrete stuff where you could test and follow along if you want. 

The main point of this part is to present the entry point of the middleware, the top level class of the code. Therefore, we will be building up to reach that goal through a Seastar introduction, the project structure and finally get to code.

# Seastar

[Seastar](https://github.com/scylladb/seastar) is an open source c++ framework designed to help build highly-performant server application. It's used by [ScyllaDb](https://www.scylladb.com/), a high-performance NoSQL Database.

One of the key feature of Seastar is it's event-driven nature which makes it easy for it's user to write non-blocking, asynchronous code. Moreover, Seastar architecture is based on the following points:

- [Shared-nothing Design](http://seastar.io/shared-nothing/): Seastar uses a shared-nothing model that shards all requests onto individual cores. 

- [Futures and Promises](http://seastar.io/futures-promises/): An advanced new model for concurrent applications that offers C++ programmers both high performance and the ability to create comprehensible, testable high-quality code. 

- [High-performance networking](http://seastar.io/networking/) : Seastar offers a choice of network stack, including conventional Linux networking for ease of development, DPDK for fast user-space networking on Linux, and native networking on OSv

- [Messaging passing](http://seastar.io/message-passing/) : A design for sharing information between CPU cores without time-consuming locking.

## A little detour 

Giving that I will be using Seastar through out this series, I thought that I should include a "How to use Seastar" section. I think it make sense to have this  detour before going further because it will make explanations of the MOM code much simpler.

I assume that you are already familiar with the concept of [Future and promise](https://en.wikipedia.org/wiki/Futures_and_promises). If you would like to dig deep into this amazing framework and learn it properly then I recommend this: [Tutorial](https://github.com/scylladb/seastar/blob/master/doc/tutorial.md), [Demos](https://github.com/scylladb/seastar/tree/master/demos) and "real world" [apps](https://github.com/scylladb/seastar/tree/master/apps).

To keep things short, We will see only basic examples and present the skeleton of Seastar programs. 

### Program Structure

Let's start with the general structure of a Seastar program:

```c++
#include <seastar/core/app-template.hh>
//other includes...

int main(int argc, char** argv) {
    seastar::app_template app;
    /*
      if you want to add option for your application
    */
    app.add_options()
        ("foo", bpo::value<std::string>(), "foo arg")
        ;
    app.run(argc, argv, [] {
            //write your logic here
            //... could retrieve args

            return /*return a future*/;
    });
}
```

Generally, every Seastar program you will write, will follow this structure: you instantiate an object of type `app_template` and then call on `run` method which takes 3 args. The last one is the most relevant for us: it's a lambda containing the logic of our application.
The `app_template` manage configuration needed by Seastar engine: logging config, CPU config, memory configs, etc... when we start our app, we can tweak those args as needed. For example, you can control the number of threads by issuing `-cNB` where NB is the number of threads you want. Note that if you choose a number higher then the number of hardware threads you have on your machine, you will get a runtime exception. I like this choice, it reduces the overhead caused by threads switches.

As for `run`, it starts the runtime(scheduler,eventloop....) that take care of executing handlers. we will discuss this later on.

### Examples

Although using Seastar brings many advantages like performance boost, Seastar asynchronous programming model comes with it's own trouble: We write code in an  unconventional way and we have to pay attention to new things. Many assumptions we hold are no longer true and this is because of Seastar asynchronous, non-blocking aspect. The following examples shows some of the issues we need to be aware of and how we can address them. 



The first example is for introduction. it just sleeps about two seconds and print out "this is not the right way to learn Seastar."

```c++
#include <seastar/core/app-template.hh>
#include <seastar/core/sleep.hh>
#include <iostream>

int main(int argc, char** argv) {
    seastar::app_template app;
    app.run(argc, argv, [] {
        using namespace std::chrono_literals;
        return seastar::sleep(2s).then([] {
            std::cout << "this is not the right way to learn Seastar.\n";
        });
    });
}

```

Note you can see the code in [compiler explorer](https://godbolt.org/z/-LmO2I).

In this program, `sleep` sets up a timer and return a future that will resolve in 2s. We chain on this future with a lambda that prints "this is.." to `std::cout`. 


The problem we have to constantly think about is object lifetime or scope. Unlike nodejs/javascript's closure, C++'s lambda doesn't spare us from dealing with object/variable accessibility. Depending on what we are doing, we can access an invalid state resulting in wrong results or we could get a runtime error. The following example demonstrates a common mistake that new users of Seastar make:

```c++
//includes ...

seastar::future<> dummy_fn(int offset) {
  int bad = 8;
  return seastar::make_ready_future<int>(offset)
      .then([&bad](int nb) { bad += nb; })
      .then([&bad] {
        bad++;
        std::cout << bad << '\n';
      });
}

int main(int argc, char **argv) {
  seastar::app_template app;
  return app.run(argc, argv, [] {
    auto ftr1 = dummy_fn(12);
    auto ftr2 = dummy_fn(46);

    return when_all(std::move(ftr1), std::move(ftr2))
        .then([](std::tuple<seastar::future<> , seastar::future<>> tf) {
            std::cout << "done computing";
            return 0;
        });
  });
}
```

You can check out the code in [Compiler Explorer](https://godbolt.org/z/RHfktS).

First let's analyze main: as usual, we have `app_template` and we call run to start Seastar _engine_ and execute the task. Next the lambda: we  call `dummy_fn` two times. This function does nothing special, it's only purpose is to show the mistake. For now, let's only keep in mind that it returns a future. The last statement start with `when_all`. This means that we will be waiting for futures `ftr1` and `ftr2` to resolve so that we can carry on with the next task(printing "computation done").

Now let's take a look at `dummy_fn`: We have a variable `bad` on which some calculation is performed. these calculation are dispatched in  lambdas chained one after another. before that, we make a ready_future(one that resolves immediately). this future resolve to an int taken from the argument `offset`. The problem is the capture by reference of `bad` : remember that we are executing asynchronously and without blocking which means we are submitting tasks to Seastar scheduler and not executing them, therefore, when the lambdas will run, `bad` will be already destroyed because we returned from `dummy_fn` and we will get wrong results.

So a general rule, if you are capturing things by reference you will mostly get in trouble unless you know what you are doing. so the default here will be capture by value.

Seastar developer thought already about this problem and supplied us with the necessary utilities. 

To correct the precedent example, we could use `do_with` function. Then `dummy_fn` becomes :

```c++
seastar::future<> dummy_fn(int offset) {
  return seastar::do_with(int{8}, [offset](int &bad) {
    return seastar::make_ready_future<int>(offset)
        .then([&bad](int nb) { bad += nb; })
        .then([&bad] {
          bad++;
          std::cout << bad << '\n';
          return seastar::make_ready_future();
        });
  });
}
```

We solve the dangling ref problem by using `do_with`. This utility manages for us the lifetime of object until we no longer need to use it. In short, it allocates memory and moves the values we will be using there then when the lamnda finish executing, it frees the memory. You should note that Seastar manages it's own memory which is pre-allocated from the start. Also note that you can tweak memory allocation size.

In addition to the previous solution, We can use reference counting to keep the variable alive until we finish computing. Using `seastar::shared_ptr`, `dummy_fn` becomes as follows:

```c++
seastar::future<> dummy_fn(int offset) {
  auto bad = seastar::make_shared<int>(8);
  return seastar::make_ready_future<int>(offset)
      .then([bad](int nb) { *bad += nb; })
      .then([bad] {
        (*bad)++;
        std::cout << *bad << '\n';
      });
}

```

Seastar offer it's own `shared_ptr` which is a non-thread safe version of `std::shared_ptr`. 
In `dummy_fn`, we make a sstar shared_ptr by calling make_shared and capture it by copy to all other lambdas. This way, we no longer have the life time problem.

Generally, we will be using `shared_ptr` but when we have a variable needed within a specific scope we may use `do_with`.

Our final example is rather oriented to expose more of Seastar constructs then to show a specific problem. With this sample, we introduce looping. Seastar offer an elegant solution for looping: `keep_doing`, `repeat`, repeat_until_value, etc.... The following is an example of using `do_until`. We get a string with `--message` option then we print it char by char:

```c++
#include <core/app-template.hh>
#include <core/sleep.hh>
#include <iostream>
#include <string>

namespace bst = boost::program_options;

int main(int argc, char **argv) {
  seastar::app_template app;
  app.add_options()("message", bst::value<std::string>(), "Message to print");
  return app.run(argc, argv, [&app] {
    auto &&config = app.configuration();
    seastar::do_with(config["message"].as<std::string>(), int{0},
                     [](std::string &msg, int &i) {
                       return seastar::do_until(
                           [&i, &msg] { return i >= msg.size(); },
                           [&i, &msg] { 
                               std::cout << msg[i] << '\n';
                               ++i;
                               return seastar::make_ready_future();
                           });
                     });
    return seastar::make_ready_future();
  });
}
```

[Compiler explorer link](https://godbolt.org/z/tc5QnT) 

We begin by adding `message` option by calling `add_options`. In main lambda, we start by getting the configuration and passing the value of `--message` option to `do_with` along with a counter initialized with 0. 

I should point out that the lambda passed to `do_with` must respect the order and the type of the other parameters: for example if we want `do_with` to manage an int and float we write: 
`do_with(int{9}, float{13.01}, [](int &arg1, float& arg2) { ...});`
 I hope I was clear on this point.

Now, let's get to the looping part, `do_until`. This function repeats a task until some condition is met. It accepts two lambdas, the first one is a predicate and the second one is the body of the loop. We check each time the value of the counter with respect to the string size. The loop end when `i` becames greater then the size of `msg`.
For the body, we are printing each char and advancing the counter.


Hopefully, we can now go back to the MOM with a basic understanding of Seastar.

# First steps

## Setting up the project

First, we are going  write cmake Files and prepare our dependency with `conanfile.txt`. I am assuming that you have installed cmake and conan So let's proceed to project structure:

![](https://thepracticaldev.s3.amazonaws.com/i/b71y7knyf3emfwhdnnrt.png)


As you can see from the picture, We have 3 main folders: one for cmake scripts, another one that contains the application it self, `src`, and a folder containing tests. 

To integrate clang-format and clang-tidy with cmake, I found these [cmake-scripts](https://git.stabletec.com/other/cmake-scripts/tree/master). note that there are other goodies there. all credits go to maintainers. copy the files into the cmake folder.

let's take care of conanfile.txt 

```
[requires]
gtest/1.8.0@lasote/stable
spdlog/1.3.1@bincrafters/stable 

[generators]
cmake
```

we are using google's lib gtests for our tests and spdlog for logging. And we are specifying cmake as our generator.

root CMakeLists.txt:

```cmake
cmake_minimum_required(VERSION 3.10)
project(Moza)

#includes cmake scripts
include(cmake/c++-standards.cmake)
include(cmake/compiler-options.cmake)
include(cmake/sanitizers.cmake)
include(cmake/formatting.cmake)
include(cmake/tools.cmake)

# require Seastar and use it's boost dependency
find_package(Seastar REQUIRED)
include_directories(${Boost_INCLUDE_DIR})

#include conan generated file for camke 
include(${CMAKE_BINARY_DIR}/conanbuildinfo.cmake)
conan_basic_setup(TARGETS)

#instruct seastar to use std::optional, std::variant etc, instead of it's own impl
add_definitions(-DSEASTAR_USE_STD_OPTIONAL_VARIANT_STRINGVIEW)

# clang format and clang-tidy integration
file(GLOB_RECURSE FFILES *.[chi]pp *.[chi]xx *.cc *.hh *.ii *.[CHI] )
clang_format(format ${FFILES})

clang_tidy(-format-style=file -checks=* -header-filter='${CMAKE_SOURCE_DIR}/*')

# we add only src folder
add_subdirectory(src)

```

Depending on the way you compiled and installed Seastar, you may need to add extras arguments for cmake so that it can find the required library. I will show how I am compiling the project later on in this article.

Next, let's look at `CMakeLists.txt` inside `src` folder:

```cmake
file(GLOB SRC_CORE lib/core/*.cpp lib/core/*.hh)


add_library(MozaLib ${SRC_CORE})
target_link_libraries(MozaLib Seastar::seastar CONAN_PKG::spdlog)


file(GLOB SRC_SERV server/*.cpp)
add_executable(Moza main.cpp ${SRC_SERV})
target_link_libraries(Moza Seastar::seastar MozaLib CONAN_PKG::spdlog)
```

We are done with preparation now so let's now get to the source folder. I structured `src` in a way that separate modules from the actual logic of the MOM: We have a `lib` sub folders containing our modules, `server` containing the source code of the class server and a `main.cpp` file.

## Main 

This is the lisiting of `main.cpp`:

```c++
#include <seastar/core/app-template.hh>
#include <seastar/core/distributed.hh>

#include "server/server.hh"

int main(int ac, char **av) {
  using namespace seastar;
  using namespace net;
  distributed<server_tcp> tcp_server;
  app_template app;
  return app.run(ac, av, [&] {
    engine().at_exit([&] { return tcp_server.stop(); });
    // here is not there
    return tcp_server.start_single().then(
        [&tcp_server] { return tcp_server.invoke_on(0, &server_tcp::start); });
  });
}
```

As evoked earlier, Seastar is an event-driven framework build around the [Reactor pattern](https://en.wikipedia.org/wiki/Reactor_pattern). Each Seastar application has a Reactor(or an event-loop) for each logical core(or hardware thread), of course assuming default configurations(no `--c` flag). These event-loop are completely independent from each other(_shared nothing design_). 

To help us shard our logic over the available core, Seastar developer introduced `seastar::distributed`. This utility take core of distributing the service(in our case server_tcp) passed as a template argument to all hardware threads.

we start by getting the local reactor instance of the executing core and attach a lambda that shutdown our service when we are leaving. `distributed` require the service to have a `stop` method. that's the only requirement.

Next, we call `start_sinlge` then invoke `server_tcp::start` on the shard(core) 0. `start_sinlge` activate only one shard.

previously I said that I wasn't using Seastar full potential. You can see that here: I am using only one shard. This is actually a mistake in my design. I didn't account for sharding from the beginning, I totlally forgot. I was stressed because deadlines are approching So even when I saw my mistake I didn't take the time to fix it even though I thought of a couple of solutions but they required changes to the protocol, something I didn't want to mess with. Hopefully, This will a topic for another post.

## Server class 

Let's focus now on the `server`, our service class. Let's get an overview before digging into the code. Note that the move and copy constructors/assignment_ops are deleted.

![](https://thepracticaldev.s3.amazonaws.com/i/nmb94rqf9gpgao9de9d2.png)



 The following is the listing of `src/server/server.hh` :

```c++
#pragma once

#include <seastar/core/reactor.hh>
#include <seastar/core/temporary_buffer.hh>
#include <seastar/net/api.hh>

#include "spdlog/spdlog.h"

#include "lib/core/connection.hh"
#include "lib/core/topology.hh"
#include "lib/protocol/parser.hh"
#include "lib/identity/identity.hh"
#include "lib/identity/identity_generator.hh"

using namespace seastar;
using namespace net;


class server_tcp {
  Topology _tp;
  lw_shared_ptr<server_socket> _listener;
  std::vector<lw_shared_ptr<Connection>> connections;
  uint16_t _port = 11212;

  std::string handleRequest(Specification sp, socket_address sco);

public:
  server_tcp() =default;
  ~server_tcp() =default;

  server_tcp(const server_tcp&) =delete;
  server_tcp(server_tcp&& ) =delete;

  server_tcp& operator=(const server_tcp&) =delete;
  server_tcp& operator=(server_tcp&&) =delete;

  server_tcp(uint16_t port) : _port{port} { }

  void start();
  future<> stop();

  future<> handleConnection(lw_shared_ptr<Connection> conn);

};
```


`Server` stores a shared ptr to a socket for listening, the port it will be listening on and an object of type Topology. let's ignore the vector of connection for now. 

Topology is the core of the server. it holds all the entitis of our Messaging model, we will discussed in more details in the next part of this series.


let's focus on `start` and `handleConnection`.

```c++
void server_tcp::start() {
  spdlog::info("server is staring...");
  listen_options lo;
  lo.reuse_address = true;
  _listener = engine().listen(make_ipv4_address({_port}), lo);

  spdlog::info("listening on port {}", _port);
  keep_doing([this] {
    return _listener->accept().then(
        [this](connected_socket fd, socket_address addr) mutable {
          auto conn = make_lw_shared<Connection>(std::move(fd), addr);
          handleConnection(conn);
        });
  });
}
```

`start` member function is simple. it makes a listener with the method `listen` of `reactor` class. we get the local instance of reactor through the call of `engine` function. Next, we start accepting connection with `accept` method from the listener and handle connections. We repeat this process using `keep_doing` utility from Seastar. Note that we are storing file descriptor(connected_socket) and the address of the client(socket_address) in a dedicated class, `Connection`. It enables  communication  with the client in both directions. let's look at the code:

```c++
//src/lib/core/connection.hh
#pragma once

#include <seastar/core/temporary_buffer.hh>
#include <seastar/net/api.hh>

using namespace seastar;
using namespace net;


class Connection {
public:
    connected_socket m_fd;
    input_stream<char> m_in;
    output_stream<char> m_out;
    socket_address m_address;
    bool m_enabled;

  public:
    Connection(connected_socket fd, socket_address addr)
      : m_fd{std::move(fd)}, m_in{m_fd.input()}, m_out{m_fd.output()}, m_address{addr}, m_enabled{true} {
      m_fd.set_keepalive(true);
      m_fd.set_nodelay(true);
    }

    Connection(Connection &&orig)
      : m_fd{std::move(orig.m_fd)}
      , m_in{std::move(orig.m_in)}
      , m_out{std::move(orig.m_out)}
      , m_address{orig.m_address}
      , m_enabled{orig.m_enabled} {
    }

    Connection &operator=(const Connection &rhs) = delete;


    Connection &operator=(Connection &&rhs) = default;

    ~Connection() {
      closeStream();
    }

    void closeStream() {
      m_enabled = false;
      m_fd.shutdown_input();
      m_fd.shutdown_output();
    }

    future<temporary_buffer<char>> readFromStream() {
      return m_in.read();
    }


    bool isAlive() {
      return m_enabled;
    }

    socket_address getRemoteAddress() const {
      return this->m_address;
    }
  };
```

The most important bits are `output_stream` and `input_stream` with which we can read/write from/to the client. We open the streams with `input()` and `output()` in `Connection` constructor.

Let's pass to `handleConnection`. This method reads from socket, parse the packet and execute the task(and finally report back). AS for now we are missing many modules and we didn't explain yet the messaging model nor the protocol, therefore, we will write it such that it reads some byte, log the request and send a response. Expect it to change when we advance in the series.

```c++
seastar::future<> server_tcp::handleConnection(lw_shared_ptr<Connection> conn) {
  return conn->readFromStream()
      .then([this](temporary_buffer<char> buf) {
        std::string str;
        std::copy(std::begin(buf), std::end(buf), std::back_inserter(str));
        spdlog::info("received data : {}", str);
      })
      .then([this, conn] {
        if (conn->isAlive()) {
          return conn->m_out.write("WYSIWYG")
              .then([conn] { return conn->m_out.flush(); })//lambda 1
              .then([conn] { return conn->m_out.close(); })//lambda 2
              .then([conn] {});//lambda 3
        }
        return make_ready_future();
      });
}
```

`readFromStream` returns a future of `temporary_buffer` passed for the next continuation in the chain when it's ready. `temporary_buffer` is an efficient container designed with the zero copy idiom and is move only. We copy the buffer into a `std::string` and log the data. Next we send a message to the client. 

Note the use of `flush` after `write`. we make sure the data is send to the client and we close the connection. 

Notice also that the last lambda in the chain is empty but it captures the `conn` by copy. This line is required to prevent an access to a destroyed object(in our case it's `m_out`): calling `conn->m_out.close();` arranges for the connection to be closed and returns a future representing that event. Excluding lambda 3 from the chain, triggers the destruction of `conn` at the end of lambda 2. However, we still need `m_out` to be alive until we finish with closing the connection. This is why we need it, even though it's empty.

For now `stop` returns a ready future :
```c++
future<> server_tcp::stop() {
  return make_ready_future<>();
}
```

Now that we finished `start` and `handleConnection` we can pass to compiling the project.


## Compiling the project

We will start by listing commands you need to type in order to compile the project then speak about some technicalties you need to be aware of. 

To compile the project first you need to make a new folder in the root directory called `build`. This contains all results of compilation, conan, etc....
```bash
mkdir build
```

we cd to `build` then type:
```bash
conan install ..
```

This will install all required package described in the `conanfile.txt`
, specifially `gtest` and `spdlog`.

Next, we build the project with cmake:
```bash
seastar_dir=/path/to/seastar
cd ..
CC=gcc-8 CXX=g++-8 cmake -DCMAKE_PREFIX_PATH="$seastar_dir/build/;$seastar_dir/build/_cooking/installed" -DCMAKE_MODULE_PATH=$seastar_dir/cmake -H. -Bbuild
```

You need to point to Seastar's cmake modules required for `find_package` and also where you installed Seastar.

I should say that you need to comment the `includes` lines from the files we worked on and the Topology attribute of  `server_tcp` before building because we didn't work on them yet.

Also, Note that I added my own profile for conan so when I run it I pass an extra command: `-pr=myProf`. The profile file is located in `~/.conan/profiles` if you keeped the default configuration when you installed conan. let's take a look at the file:
```
[settings]
os=Linux
arch=x86_64
compiler=gcc
compiler.version=8.1
compiler.libcxx=libstdc++
build_type=Release
[options]
[build_requires]
[env]

```


I build Seastar using the `cooking.sh` way from their [how to](https://github.com/scylladb/seastar/blob/master/HACKING.md) on building Seastar. 

# Conclusion
In this post we learned how to use Seastar and we wrote the first building blocks of our project. I hope you have enjoyed reading.


# Next Part

Next on this series, We will discuss the Messaging model of our system, the core of MOM.